Flink 1.14.6中动态表与版本化表左连接的最优方案咨询

需要将两个流以左连接方式关联，为左侧流补充右侧流信息：

• 左侧流：car_traffic
• 右侧流：car_electronics
• 关联字段：license_plate_number

对car_electronics的处理要求：仅保留license_plate_number和gps_mac_addr字段，过滤非空值后转为版本化表视图（MAC地址会频繁变更，部分车辆无GPS模块）。核心需求是用右侧的GPS MAC地址参考表 enrich 左侧流，同时保留左侧无匹配的记录。

数据规模：日处理量16-20亿条记录，使用Flink 1.14.6版本。

咨询以下两个问题：

1. 关联这两个流的最佳实现方式是什么？
2. 应选用哪种连接类型？

数据示例 #

car_traffic 流 #

+------------------------+--------------------------+----------------------+
|   license_plate_number |                eventTime |                  ... |
+------------------------+--------------------------+----------------------+
|              AA-123-BB |  2022-11-29          ... |                  ... |
|              AA-456-CC |  2022-11-29          ... |                  ... |
|              EE-935-JJ |  2022-11-29          ... |                  ... |

car_electronics 流 #

+----+----------------------+-------------------+--------------------------+
| op | license_plate_number |      gps_mac_addr |                eventTime |
+----+----------------------+-------------------+--------------------------+
| +I |            AA-123-BB |                AA |  2022-11-28          ... |
| -U |            AA-123-BB |                AA |  2022-11-29          ... |
| +U |            AA-123-BB |      FFFF0A0FBBC6 |  2022-11-29          ... |
| +I |            AA-456-CC |      FFFF0A0F00F0 |  2022-11-29          ... |

期望结果 #

+------------------------+------------------------+------------------------+
|   license_plate_number |           gps_mac_addr |              eventTime |
+------------------------+------------------------+------------------------+
|              AA-123-BB |           FFFF0A0FBBC6 |                    ... |
|              AA-456-CC |           FFFF0A0F00F0 |                    ... |
|              EE-935-JJ |                 (NULL) |                    ... |

1. 最佳实现方式 #

基于你的需求和Flink 1.14.6版本，推荐采用Flink SQL/Table API结合版本化维表的方案，具体步骤如下：

步骤1：预处理右侧流为版本化维表 #

先对car_electronics流做清洗，保留目标字段并过滤空值，再将其定义为版本化维表，基于eventTime维护每个license_plate_number的最新版本：

-- 定义car_electronics流表，适配CDC格式的op字段
CREATE TABLE car_electronics_dim (
    license_plate_number STRING PRIMARY KEY NOT ENFORCED,
    gps_mac_addr STRING,
    eventTime TIMESTAMP(3) METADATA FROM 'value.eventTime',
    WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka', -- 根据实际数据源调整
    'topic' = 'car_electronics_topic',
    'format' = 'debezium-json',
    'scan.startup.mode' = 'latest-offset'
);

-- 创建过滤非空值的视图
CREATE VIEW car_electronics_valid AS
SELECT license_plate_number, gps_mac_addr, eventTime
FROM car_electronics_dim
WHERE gps_mac_addr IS NOT NULL;

步骤2：时态左连接 enrich 左侧流 #

将car_traffic流与预处理后的维表做时态左连接，自动关联每个车牌对应的最新GPS MAC地址：

-- 定义car_traffic流表
CREATE TABLE car_traffic (
    license_plate_number STRING,
    eventTime TIMESTAMP(3) METADATA FROM 'value.eventTime',
    -- 其他业务字段
    WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka', -- 根据实际数据源调整
    'topic' = 'car_traffic_topic',
    'format' = 'json'
);

-- 执行左连接获取结果
SELECT 
    t.license_plate_number,
    COALESCE(d.gps_mac_addr, NULL) AS gps_mac_addr,
    t.eventTime,
    -- 左侧流其他业务字段
FROM car_traffic t
LEFT JOIN car_electronics_valid FOR SYSTEM_TIME AS OF t.eventTime d
ON t.license_plate_number = d.license_plate_number;

性能优化要点 #

针对日处理16-20亿条的规模，需注意：

• 根据集群CPU资源设置并行度，建议为核心数的2-3倍
• 开启状态后端的增量Checkpoint，降低快照开销
• 为license_plate_number的状态设置TTL（例如7天），避免状态无限膨胀
• 调整Kafka消费者参数（fetch.min.bytes、fetch.max.wait.ms），平衡吞吐量与延迟

2. 应选用的连接类型 #

必须选用基于EventTime的时态左连接（Temporal Left Join），原因如下：

• 完全满足左连接需求：保留car_traffic中所有无匹配的记录，对应gps_mac_addr为NULL
• 适配右侧流的版本化特性：自动跟踪每个车牌的最新GPS MAC地址，确保左侧流记录关联到对应时刻的最新数据
• 适配流处理时序特性：基于EventTime的水位线机制，能处理乱序数据，保证关联结果的正确性

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