Need some help understanding how to approach reprocessing historical data. Suppose we have S3 buckets A, B, and C with time series data. All 3 buckets are in Parquet format, partitioned by day, vary in volume, and have a common primary key and event time. Goal Execute a job that joins data from each bucket by the primary key such that at a given moment (millisecond precision), the job sees only the latest data from each bucket based on event time (watermark). In essence, the job needs to time travel as if it only had the latest data from A, B, and C for each past moment. This would seem possible in Flink but here's where I'm stuck: • According to the documentation, "When using a directory as the source path, there is no defined order of ingestion for the files inside the directory." which suggests events won't be processed in order in a partition dir. For example, "part-13.parquet" could be read before "part-5.parquet" in dir "dt=2022-10-01". This would seem to prevent a job from processing a bulk format stream in event time order. • Will the different volumes of the S3 data create an issue? Suppose bucket A had 4x more data than bucket B - will the job still be able to look at the latest rows from A, B, and C for a given moment when bucket B is processed much quicker than A?

• According to the documentation, “When using a directory as the source path, there is no defined order of ingestion for the files inside the directory.” which suggests events won’t be processed in order in a partition dir. For example, “part-13.parquet” could be read before “part-5.parquet” in dir “dt=2022-10-01". This would seem to prevent a job from processing a bulk format stream in event time order.

That’s correct, you need a custom

FileSplitAssigner

with some tweaks in the

FileSource

(I really need to create a JIRA issue for that…) if you want to read files in order

@sap1ens thank you - makes sense. If a custom FileSplitAssigner were to process the dir in order, would I be able to achieve proper time travel with the 3 bulk format sources?

Hard to say. Might be doable with very large late arriving time config? 🤷

Hmm, ok. Given the sizes and varying ingestion rates of A, B, and C, that would seem hard to achieve. Not sure how Flink would behave with a 1 month buffer period, for example. One idea I was thinking about is to run a 1-time batch job to create a new dataset that combines the original datasets into a single dataset ordered by event time. Would that work in theory to properly time travel through A, B, and C datasets?

I am just commenting here to be alerted of all future discussions here. I have this exact issue