使用Flink Table API进行多表关联的状态创建、底层状态形态及并行场景数据重排机制咨询

Let’s break down your questions one by one, with clear explanations of what’s happening under the hood.

1. 多表关联时Flink创建的状态类型 #

When executing multi-table joins with the Flink Table API, Flink creates several types of state depending on the join type and time semantics:

• Keyed Join Buffer State: This is the core state for stream joins. For each side of the join, Flink uses keyed state (usually MapState or ListState) to cache records that haven’t yet found a matching counterpart from the other stream. The state is partitioned by the join key, so only records with the same key are stored together. For example, in an inner join between A and B, Flink will cache A’s records by their pk_id and B’s records by their fk_id (which maps to A’s pk_id) until a match is found.
• TTL & Watermark State: If you’re using event time, Flink maintains state to track watermarks and enforce time-to-live (TTL) policies for join buffers. This prevents state from growing indefinitely by cleaning up records that are too old to ever find a match (based on watermark progress). For processing time joins, TTL is still used but based on wall-clock time instead of event time.
• Window State: If your join is constrained to a window, Flink creates window-specific state to store records within each window. Once a window closes, the state for that window is cleaned up after the join is processed.
• Outer Join Retention State: For left/right/full outer joins, Flink needs to retain records from one or both sides (until TTL expires) to ensure that even if a matching record arrives later, the join result can still be emitted. This state is more persistent compared to inner join buffers, which may discard records once a match is found.

2. 三表内关联的底层状态形态与并行重排 #

Let’s take your example query:

select * from A inner join B on a.pk_id = b.fk_id inner join C on b.pk_id = c.fk_id

底层状态形态 #

Flink doesn’t perform a single three-way join directly—it breaks the query into a two-step sequential join pipeline:

1. First step: A ⋈ B

   • Flink partitions both streams A and B by their join key (a.pk_id for A, b.fk_id for B, which are equivalent per the join condition).
   • For each parallel subtask, it maintains two keyed state buffers:
     • One buffer for A’s records that haven’t found a matching B record yet, keyed by pk_id.
     • Another buffer for B’s records that haven’t found a matching A record yet, keyed by fk_id.
   • When a record arrives from A, Flink checks the B buffer for matching entries. If found, it emits the joined AB record and cleans up any expired entries from both buffers. If no match is found, the A record is added to its buffer. The same logic applies to incoming B records.

2. Second step: AB ⋈ C

   • The intermediate AB stream (containing fields from A and B) is joined with C using b.pk_id = c.fk_id.
   • Flink partitions the AB stream by b.pk_id and the C stream by c.fk_id (which maps to b.pk_id).
   • Each parallel subtask maintains two keyed state buffers:
     • One for AB records waiting for a matching C record, keyed by b.pk_id.
     • One for C records waiting for a matching AB record, keyed by c.fk_id.
   • Matching logic mirrors the first step: when a matching pair is found, the final joined record (Z) is emitted, and expired entries are cleaned up.

并行运行时的数据重排 #

Yes, Flink will definitely shuffle data when running in parallel—this is mandatory for distributed joins. Here’s why:

For a join to work correctly, all records with the same join key must be processed by the same parallel subtask. If the original streams are partitioned differently (e.g., A is partitioned by some other field, not pk_id), Flink will perform a hash shuffle:

• For the A ⋈ B step: A’s records are re-partitioned by a.pk_id, B’s records by b.fk_id—since these keys are equal, they’ll hash to the same subtask.
• For the AB ⋈ C step: AB’s records are re-partitioned by b.pk_id, C’s records by c.fk_id—again, ensuring same-key records land in the same subtask.

Even if the source streams are already partitioned by the join keys, Flink may still validate or re-partition them to ensure consistency across the job graph. This shuffle is a core part of distributed stream processing for joins—without it, records with matching keys could end up in different subtasks and never be joined.

内容的提问来源于stack exchange，提问作者urield