A presentation at Flink Forward 2025 in in Barcelona, Spain by Gustavo de Morais
The “Big State” Monster: Taming State Size in MultiWay Joins with FLIP-516 Gustavo de Morais, Staff Software Engineer@Confluent 13-16 October 2025 Barcelona 2025
Table of Contents 1. What are multi-way joins? 2. Why are they problematic? 3. How can we address that? The MultiJoin Operator 4. Benchmark: Binary Joins vs MultiJoin 5. What types of joins are supported? 6. Optimizations and next steps
What are multi-way joins? A big challenge for stateful scalable distributed stream processing
What are multi-way joins? select * from users join orders on users.id = orders.user_id join shipments on shipments.orderid = orders.id join details on shipments.id = details.sid join … on …
Multiple chained non-temporal binary joins users Join orders shipments details Join Join Sink
Multiple chained non-temporal binary joins users Join orders shipments details Join Join Sink
Why are they problematic?
Why are they problematic? users Join Operator event reaction 1 event 1 reaction 1 user x 1 event x 1 reaction 1 user x 1 event 1 user Join Operator Join Operator
Why are they problematic? users 100000 users Join Operator 100000 users x 1000 events 10 M event reaction 1000 event 10 reaction 100000 users x 1000s event x 10s reaction Join Operator Join Operator 100 M records
Why are they problematic? users 100000 users Join Operator 100000 users x 1000 events 10 M event 100000 users x 1000s event x 10s reaction Join Operator Join Operator 100 M records 1000 event 🚨 We store all of them in state! reaction 10 reaction ~100k records ~10M records ~100M records
Why are they problematic?
Why are they problematic? ● State and checkpoints size can become huge ● Checkpoints start to timeout ● Recovery time is long due to state size ● Processing rate drops heavily ● … and the pipeline becomes unhealthy
How can we address that? FLIP-516 Multi-way Join Operator
FLIP-516: Multi-way Join Operator users orders shipments MultiJoin Operator Sink details
FLIP-516: Multi-way Join Operator users orders ✅ Only input records in state ✅ No intermediate state shipments MultiJoin Operator ✅ Faster for big jobs ✅ No shuffling 🟡 Data skew ⛔️ Reprocessing details ⛔️ Common join key required
FLIP-516: Multi-way Join Operator users 100000 user Join Operator 100000 users x 1000 events 100000 users x 1000s event x 10s reaction Join Operator Join Operator 10 M event reaction 100 M records 1000 event 10 reaction We use ~1% of the previous state ~100k records ~10M records ~100M records
Why are they problematic?
FLIP-516: Multi-way Join Operator SET ‘table.optimizer.multi-join.enabled’ = ‘true’;
Benchmark Binary Joins vs MultiJoin
Benchmark: parameters ● Tried to keep as generic as possible ● Parallelism 10 ● Records equally distributed ● 10 chained LEFT joins ● No considerable record amplification ● RocksDB as state backend ● 2.8 GB Managed Memory (used for state cache) ● Streaming mode ● Running in Apple M1 Pro 32 GB
Benchmark: Execution Time vs Input Records
Benchmark: State Size vs Input Records
What types of joins are supported?
What types of joins are supported? Supported ✅ Inner joins Future Work ❌ FULL OUTER JOINS - also requires calcite change ✅ Left joins 🚧 Right joins (wip) ❌ SEMI and ANTI joins - also requires calcite change ✅ Streaming mode ❌ Temporal Joins
Partitioning: common join key Supported Not supported ✅ A JOIN B ON A.key1 = B.key JOIN C ON A.key1 = C.key ❌ A JOIN B ON A.key1 = B.key1 JOIN C ON B.key2 = C.key2 ✅ A JOIN B ON A.key = B.key1 JOIN C ON B.key1 = C.key2 (Partition by key via transitivity) ✅ A JOIN B ON A.key = B.key1 JOIN C ON B.key1 = C.key2 AND B.key10 = C.key.10
Partitioning: common join key MultiJoin Operator Color? MultiJoin Operator MultiJoin Operator
Partitioning: common join key MultiJoin Operator Color? MultiJoin Operator ? ? ? ? MultiJoin Operator
Partitioning with multiple keys MultiJoin Operator MultiJoin Operator
Optimizations and next steps
Optimizations and next steps ● Set STATE TTL ○ Try to keep the state size < Managed Memory ● Stick to append mode ● Avoid outer joins ○ Toggle upsert mode ○ Have natural amplification ● Try to preserve upsert keys ● Optimize binary joins ○ Reorder Joins ○ Less frequently updated tables to the left -> More frequently updated tables to the right ● Use temporal joins
Optimizations and next steps ● Stabilize MultiJoinOperator for Flink 2.2 ● (Adaptive) MultiJoinOperator ○ Cache some intermediate state ○ Cache hot keys ○ Change traversal order ● BroadcastMultiJoinOperator ● DeltaMultiJoinOperator ● LookupMultiJoin ● Temporal MultiJoinOperators
More details Documentation
Thank You Gustavo de Morais, Staff Software Engineer at Confluent Barcelona 2025
Extra slides Add Title of the Upcoming Section
Operator State Layout Extra slide for questions
Operator State Layout JoinKeyContainsUniqueKey ● MapStateDescriptor<RowData, RowData> ● Stores data as MapState<JoinKey, Record> InputSideHasUniqueKey ● MapState<RowData, RowData> ● Stores data as MapState<CompositeKey<JoinKey, UniqueKey>, Record> InputSideHasNoUniqueKey ● MapStateDescriptor<RowData, Integer> ● Stores data as MapState<CompositeKey<JoinKey, Record>, Count>
FLIP-516: Multi-way Join Operator What if we have record amplification? Docs: https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/tuning/#multijoin-operator-example—-benchmark
Benchmark: Binary Joins vs MultiJoin
MultiJoin Operator Inside the operator - examples
Inside the StreamingMultiJoinOperator SELECT users.id, users.name, orders.id, payment.status FROM users JOIN orders on users.id = orders.uid JOIN payment orders.uid = payment.uid AND orders.id = payment.oid users orders [1, “Joe”] [1, “order_1”, “Phone”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] ️1 payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “Car”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] output payments [1, “order_1”, “paid”] ❌ +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_2”, “Car”] payments [1, “order_1”, “paid”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Phone”, “paid”] ❌
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator [1, “order_2”, “pending”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Phone”, “paid”] ✅ +I[1, “Joe”, “Car”, “pending”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”] ❌
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”] ✅
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”] ✅ +U[1, “Joe”, “Car”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] output payments users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “pending”] +I[1, “Joe”, “Car”, “pending”] [1, “order_2”, “paid”] ✅ +U[1, “Joe”, “Car”, “paid”]
The new StreamingMultiJoinOperator +U[1, “order_2”, “paid”] payments output users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “paid”] +I[1, “Joe”, “Phone”, “paid”] [1, “order_2”, “Car”] [1, “order_2”, “paid”] +I[1, “Joe”, “Car”, “pending”] +U[1, “Joe”, “Car”, “paid”]
Inside The new theStreamingMultiJoinOperator StreamingMultiJoinOperator - Left join SELECT users.id, users.name, orders.id, payment.status FROM users LEFT JOIN orders on users.id = orders.uid LEFT JOIN payment orders.uid = payment.uid AND orders.id = payment.oid users orders payments output
The new StreamingMultiJoinOperator - Left join +I[1, “Joe”] users orders ️1 payments output
The new StreamingMultiJoinOperator - Left join +I[1, “Joe”] users +I [1, “Joe”] orders payments output +I[1, “Joe”, “Phone”, “paid”]
The new StreamingMultiJoinOperator - Left join [1, “Joe”] orders users +I [1, “Joe”] 🔗0 payments output
The new StreamingMultiJoinOperator - Left join [1, “Joe”] orders users +I [1, “Joe”] 🔗0 [null, null, null] payments output
The new StreamingMultiJoinOperator - Left join [1, “Joe”] orders users +I [1, “Joe”] 🔗0 [null, null, null] payments output
The new StreamingMultiJoinOperator - Left join [1, “Joe”] +I [1, “Joe”] payments orders users 🔗0 [null, null, null] 🔗0 output
The new StreamingMultiJoinOperator - Left join [1, “Joe”] +I [1, “Joe”] payments orders users 🔗0 [null, null, null] 🔗0 [null, null, null] output
The new StreamingMultiJoinOperator - Left join [1, “Joe”] +I [1, “Joe”] payments orders users 🔗0 [null, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “Joe”] +I [1, “Joe”] payments orders users 🔗0 [null, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “Joe”] orders users +I [1, “Joe”] 🔗0 [null, null, null] payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “Joe”] users +I [1, “Joe”] orders payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join Waiting for records users +I [1, “Joe”] orders payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone”] users +I [1, “Joe”] orders payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] +I users orders [1, “Joe”] [1, “order_1”, “Phone”] payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] +I users orders [1, “Joe”] [1, “order_1”, “Phone”] payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] orders users +I [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] ✅ output +I[1, “Joe”, null, null] Important transition
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗1 [null, null, null] payments output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗1 [null, null, null] payments output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] orders users +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone”] payments output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] 🔗0 [null, null, null] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] 🔗0 [null, null, null] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “paid”] orders users +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone”] payments output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “paid”] +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone”] output payments orders users 🔗1 [1, “order_1”, “paid”] ✅ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “paid”] -D [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] 🔗1 [null, null, null] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null]
The new StreamingMultiJoinOperator - Left join [1, “order_1”, “paid”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] 🔗1 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] +I[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I[1, “order_1”, “Phone 1”] users orders [1, “Joe”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] +I[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] users orders [1, “Joe”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] users orders [1, “Joe”] [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] ✅ [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] [1, “order_1”, “Phone”] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, This is a essential reprocessing
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] -D [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] [1, order_1, paid] ✅ output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
West Eleven Restaurant The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] -D [1, “Joe”] payments orders users 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] [1, order_1, paid] ✅ output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] output payments [1, order_1, paid] ✅ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1,
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] output payments [1, order_1, paid] ✅ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] output payments [1, order_1, paid] ✅ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users +I [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [null, null, null] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗0 [1, “order_1”, “Phone”] [null, null, null] 🔗0 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [null, null, null] output payments orders users 🔗0 [1, order_1, paid] ❌ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗0 [1, “order_1”, “Phone”] [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗0 [1, “order_1”, “Phone”] [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] +I [1, “Joe”] payments orders users 🔗0 [1, “order_1”, “Phone”] [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 [null, null, null] output +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] orders users -D [1, “Joe”] 🔗0 [1, “order_1”, “Phone”] [null, null, null] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] -D users orders [1, “Joe”] [1, “order_1”, “Phone”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join -D[1, “order_1”, “Phone”] users -D [1, “Joe”] orders payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] Updated state
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] users [1, “Joe”] orders payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] users orders [1, “Joe”] [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] users orders [1, “Joe”] [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users +I [1, “Joe”] 🔗0 [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone 1”] ✅ [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users -D [1, “Joe”] 🔗1 [null, null, null] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] 🔗0 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] -D [1, “Joe”] 🔗1 [null, null, null] output payments orders users 🔗0 [1, order_1, paid] ❌ +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] output [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] output [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] -D [1, “Joe”] payments orders users 🔗1 [null, null, null] output [1, order_1, paid] +I[1, “Joe”, null, null] [null, null, null] -D[1, “Joe”, null, null] 🔗0 +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users -D [1, “Joe”] 🔗1 [null, null, null] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone 1”] 🔗0 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone 1”] 🔗1 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone 1”] 🔗1 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, paid] order.name field is now Phone
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] +I [1, “Joe”] payments orders users 🔗1 [1, “order_1”, “Phone 1”] 🔗1 [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join +I[1, “order_1”, “Phone 1”] orders users +I [1, “Joe”] 🔗1 [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, paid]
The new StreamingMultiJoinOperator - Left join Waiting on new events users orders [1, “Joe”] [1, “order_1”, “Phone 1”] payments [1, order_1, paid] output +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, null] -D[1, “Joe”, order_1, null] -I[1, “Joe”, order_1, paid] +I[1, -D[1, “Joe”, order_1, paid] +I[1, “Joe”, null, null] -D[1, “Joe”, null, null] +I[1, “Joe”, order_1, paid]
Wave PatternsFeel free to integrate into your presentation
2009 2009 2009 2009 2009 2009
Real-time data is essential, but often the path to it is fraught with hidden challenges. One of the biggest culprits in complex stream processing applications, particularly when dealing with multiple data sources, is the “big state problem.” When you chain together several binary joins in Apache Flink, each intermediate result needs to be stored in state, leading to an ever-growing monster that devours resources and degrades jobs.
Key takeaways from this session will include:
Understanding the “Big State” Problem: Why chaining binary joins leads to excessive state. Introducing the StreamingMultiJoinOperator: How it works to reduce state. Realized Benefits: Exploring the impact of zero intermediate state and new possible optimizations. Practical Considerations: Discussing current limitations and compatibility. Other approaches: What are other strategies on dealing with large joining jobs? What’s next for multi-way joins in Flink. By the end of this session, you’ll gain a clear understanding of how to mitigate the “big state” problem and unlock the full potential of efficient, high-performance multi-way joins in your Apache Flink streaming applications. The “big state” monster will be far less scary, and your real-time data journey will be smoother than ever!”
