开发者社区大数据与机器学习正文

Flink CDC 能保证执行顺序吗？比如我订单表有流过来了，但是商品表流还没过来，导致订单找不到商

Flink CDC 能保证执行顺序吗？比如我订单表有流过来了，但是商品表流还没过来，导致订单找不到商品，这种情况怎么解决？我把商品表的cdc放在订单表的cdc上面，可以解决这个问题吗？cdc也需要时间吧我怎么保证商品表全部加载过来了呢？

展开

收起

真的很搞笑 2023-06-04 18:14:13 90 版权

阿里云 AI 助理回答

Flink CDC 本身无法直接保证多个表之间的执行顺序，因为每个表的变更数据捕获（CDC）是独立进行的。如果订单表的流数据先到达，而商品表的数据尚未完全加载，可能会导致订单无法找到对应的商品信息。这种问题通常需要通过以下方法来解决：

1. 使用维表 Join 解决依赖问题

在 Flink 中，可以通过维表（Dimension Table）的方式将商品表作为维表与订单表进行关联。这种方式可以确保即使商品表的数据稍后到达，订单表仍然能够正确地找到对应的商品信息。

实现方式：
- 将商品表注册为维表，并设置缓存策略（如全量缓存或部分缓存）。
- 在订单表的流处理逻辑中，通过 JOIN 操作将订单表与商品维表关联。
- 如果商品表的数据更新频率较低，可以选择全量缓存；如果更新频率较高，则可以使用 TTL 缓存策略。
注意事项：
- 维表的缓存策略需要根据业务需求进行调整，以平衡性能和数据一致性。
- 如果商品表的数据量较大，建议使用分布式存储（如 Hologres 或 Elasticsearch）作为维表的存储介质。

2. 调整 CDC 数据同步顺序

虽然将商品表的 CDC 放在订单表的 CDC 上方可以在一定程度上缓解问题，但这并不能完全保证商品表的数据已经全部加载完成。CDC 的全量阶段可能需要一定时间，尤其是在数据量较大的情况下。

如何确认商品表已加载完成：
- 监控指标：通过 Flink 的监控指标 currentEmitEventTimeLag 来判断商品表是否已完成全量数据同步。当该指标大于 0 时，表示全量数据同步已完成，进入增量读取阶段。
- 日志检查：在 MySQL CDC 源表所在的 TaskManager 日志中查找 BinlogSplitReader is created 日志，确认商品表的全量数据是否已读取完成。
优化建议：
- 如果商品表的数据量较大，可以通过并发读取全量数据来加速加载过程。例如，Postgres CDC 和 MongoDB CDC 都支持并发读取全量数据。
- 对于 MySQL CDC，可以通过配置 scan.incremental.snapshot.enabled=true 参数启用增量快照算法，从而提高全量数据读取的效率。

3. 使用双流 Join 并设置延迟容忍

如果商品表和订单表都需要实时处理，可以使用双流 Join 的方式，并设置一定的延迟容忍时间（Watermark 和 Allowed Lateness）。

实现方式：

为商品表和订单表分别定义 Watermark，用于标记事件时间。
设置 Allowed Lateness，允许订单表在商品表数据稍后到达的情况下仍然能够正确关联。
示例代码如下：

SELECT 
  o.order_id,
  o.product_id,
  p.product_name
FROM 
  order_stream AS o
JOIN 
  product_stream AS p
ON 
  o.product_id = p.product_id
AND 
  o.event_time BETWEEN p.event_time - INTERVAL '5' MINUTE AND p.event_time + INTERVAL '5' MINUTE

注意事项：
- 双流 Join 的性能会受到状态大小的影响，因此需要合理设置状态 TTL，避免状态过大导致性能下降。
- 如果商品表的数据更新频率较低，可以考虑将其作为维表而非流表处理。

4. 数据预加载与初始化

在某些场景下，可以通过预加载商品表的数据来确保其在订单表处理之前已经可用。

实现方式：
- 在作业启动前，手动将商品表的全量数据加载到目标存储（如 Hologres 或 Kafka）中。
- 启动作业后，订单表的流数据可以直接与预加载的商品数据进行关联。
注意事项：
- 预加载的方式适用于商品表数据变化较少的场景。如果商品表数据频繁更新，仍需结合 CDC 实现增量同步。