Really trying to understand how to use flink here, and so far failing. I have an existing algorithm, written in pandas code that looks at a series of data points and enriches them using pandas. It then returns a cleaned up data set which removes outliers and accounts for discontinuity among other things. I cannot seem to get this to work inline in pyFlink where I can do:

df = table.to_pandas()
clean_up_data_(df)
clean_table = t_env.from_pandas(df)

I can't really find any documentation on this use case, and it seem like doing this in a udf might be the way to go, but even that seems to only get multiple columns of a table. Ideally i'm doing this over a time window of data and only emitting the clean data.

Hey, hopefully you find this helpful - if you’re looking at each data point as an individual element/row/object - then you can apply your time filter to “allow” the elements that fit to go “down stream.” So that’s something like: intitialDataStream.filter(thing where thing.time is < something)

But if I write a udf i'm not receiving a data frame it seems to be a single row

Generally in Java-land we may use a process function and only tell the context to collect after some “complex” logic

That would be ideal, I think there is little appetite to rewrite the legacy code

It really depends, my past experience - enrichment would be an HTTP API call or database look up.

Got it