@gmodena, I've been trying to write tests for eventutilities-python, so that we could more easily improve and add things (like error event side output, etc.).

I've spent two days struggling with writing and testing a simple enrichment pipeline. I've even simplified and am just trying to write a simple python datastream enrichment map function that goes from one stream with a schema to an output with a different schema. I could not get this to work!

It took me forever to realize how this works for the page content enrich pipeline: the output is the same schema as the input! If the output schema is different, we can't just treat the Row in the map function as a dict. Doing so will result in a ValueError being thrown when trying to assign a field to Row that the Row doesn't already know about.

I tried to work around this by recursively converting the Row to a dict before providing it to the map function. This works fine, but we have no way to convert a nested dict back to a Row recursively (unless we write it ourselves). (I tried RowTypeInfo.from_internal_type, but I couldn't get it to work?).

I think if we want to be able to treat the event like a Python dict, we are going to have to implement custom recursive converters in Python between pyflink Row <-> and python dict.

I'd hope we can do somethign like this:

def enrich_fn(event: dict) -> dict:
    event['enriched_field'] = 'enriched value'
    return event

# gets the source and sink row types via EventDataStreamFactory
with stream_manager(source_stream_name='...', sink_stream_name='...', sink='kafka...?') as stream:
    stream.map(enrich_fn)
    stream.execute()

To do this, the functions users implement all have to work with dicts, which will require conversion of the input data stream, and also conversion of the final output datastream back to Row of the output RowTypeInfo.

Okay, I think I have something working?

We can already easily recursively convert Rows to dicts.

dict_to_row will recursively convert any dicts that should be Rows to Rows using the RowTypeInfo.

This allowed me to make the stream manager stuff always pre-convert the datastream to dicts, and then convert back to the output Row type before sending to the sink.

In this way, all user provided map, filter, etc. functions work with dicts.

Still SUPER WIP, but I got it working with our Kafka sink stuff, and was able to output using a different schema.

In T324689#8500218, @Ottomata wrote:

It took me forever to realize how this works for the page content enrich pipeline: the output is the same schema as the input! If the output schema is different, we can't just treat the Row in the map function as a dict. Doing so will result in a ValueError being thrown when trying to assign a field to Row that the Row doesn't already know about.

I tried to work around this by recursively converting the Row to a dict before providing it to the map function. This works fine, but we have no way to convert a nested dict back to a Row recursively (unless we write it ourselves). (I tried RowTypeInfo.from_internal_type, but I couldn't get it to work?).

I think if we want to be able to treat the event like a Python dict, we are going to have to implement custom recursive converters in Python between pyflink Row <-> and python dict.

Correct, this is not implemented yet. Last quarter we ended up prioritising the rollout of a python version of mediawiki stream enrichment over API completeness.
Let's address this missing bit ASAP though. We'll need it, among other things, to produce messages to an error topic.

I was wondering if we could reuse the projection/row creation primitives from JVM eventutilities, but it turned out that would require additional (non trivial) wrapping. Doing the conversion in Python should be more straightforward.

In T324689#8503145, @Ottomata wrote:

Okay, I think I have something working?

Terrific!

LGTM. Maybe we can test it out within the scope of https://phabricator.wikimedia.org/T326536?
Would be great to have some unit tests (more in general), but it's tricky with the current CI setup / java deps. It may be be time
to fix that properly.

I do wonder how much overhead this conversion will introduce. It should not be too bad (especially at low throughput),
but something we might want to instrument.

gmodena updated https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/-/merge_requests/11

Add link to WIP documentation

gmodena merged https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/-/merge_requests/11

Add link to WIP documentation