Hey all. We're working on transitioning our batch ETLs infrastructure to streaming using Flink. One of our stream jobs is written in Flink SQL. It's a query with probably 15 joins (inner and left). We're troubleshooting performance, missing data, and scale issues... First, trying to understand how to joins work with partitioning. How kafka partitions relate to flink joins, etc... 1. Do all the topics that we're joining on need to be partitioned on the same key and data collocated? I can't seem to find this info for Flink, but found information about this for KSQL which seems to state this. 2. If data isn't collocated and partitioned consistently across all join topics, does that mean we will be missing data? 3. How does a Flink SQL job with parallelism work with partitioned topics? Can someone shed some light onto this for us and/or point to some documentation/tutorials?

Hey Ilya, Flink can repartition the data as needed for a join so you are not bound by the partition of the kafka topic. When you join two data streams by key, Flink will partition the data on that key and spread it across the flink workers for parallelism (e.g. if you are joining on customerid, all records with customerid=5 will be sent to the same Flink worker for joining). See https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/queries/joins/ for more details on joins. It's important to understand what type of join you are executing - specifically if you are joining within a time window, temporal join, or inner join. This is the most likely cause of "missing" data: if you are joining within a time window but the watermarks or timestamps are not configured correctly, it can lead to missing data.

Thanks Matthias. We're not joining within time windows. They are straight inner joins. But we're seeing constant issues with out of resource failures and then missing data.

Hey Ilya, I am not surprised that you are seeing performance issues. 15 inner joins is a lot. When Flink executes an inner join in streaming mode, it has to hold both "sides" of the join in memory and it accumulates over time. Is there a way to convert them to time-window or temporal joins? Those are executed much more efficiently. As you transition from batch to streaming, that may require a little bit of "rethinking" of your query. I think going that route will be more beneficial for you long term than trying to performance optimize your Flink setup. But that's just my 2 cents.

Hey Matthias, probably not, just given that we're doing CDC and it would require us to fix a lot of upstream stuff we (our group) don't currently control. I'm not sure about time windows either, because most of the topics we join are are core and are valid over long periods of time. Say we have study, site, patient, event, etc... When data comes in for that patient, have to join that datapoint to all the higher parents. If we had a true event driven architecture, we'd publish these events with all the context. This join is our way to create this event, but by joining all the tables.

Hey @Ilya Sterin, intermediate jobs will make it easier to operate and can give you better insight into what is going on. But the combinatorial explosion of multi-way joins may still become an issue and you may end up kicking the can just a little further down the road. Maybe temporal joins can help you here. Let's say you have an "adverse drug reaction" event with the schema (event_id, timestamp, event_type, event_message, patient_id) and you want to join it with the patient table, the patient-site-assignment table, the site table, the site-study-assignment table, and the study table. As an inner join, this 6 way join can become a memory issue. As a temporal join it won't be an issue because you are joining the event with the other dimensional tables at the point in time of the event which allows Flink to free up memory quickly. For context augmentation, temporal joins are the way to go imho. DataSQRL might make this a lot easier for you. It's a open-source compiler for Flink that distinguishes between stream/event and state tables and automatically picks the best join type. It also makes it easy to export intermediate results for reuse if you want to break up that massive join you are working on. But you can also do this directly in Flink.