Flink SQL 做流上数据打宽，这几年已经算是实时数仓里的常规操作了。但我在不少交流群和线下分享里发现，很多人对 Flink SQL Join 的理解还停留在“写个 JOIN 语法就能跑”的阶段，真正遇到数据倾斜、状态膨胀、乱序补偿、延迟关联这些问题时，往往手足无措。这篇文章我就结合自己实际做过的一个 FlinkSQL-Join 打宽案例，把从需求拆解到方案落地、再到问题排查的完整过程都摊开来讲，希望能给正在做实时数仓或正在琢磨 Flink SQL 的同学一些可以“抄作业”的参考。

先说下这个案例的背景。当时业务方需要一个实时订单大宽表，把订单主表、用户维表、商品维表、以及另一路支付流水流进行关联，最终输出包含完整用户信息、商品信息、支付信息的宽表数据，供下游实时报表和数据分析使用。整个链路如果用传统的 Lambda 架构去做，离线 T+1 的大宽表只能处理历史数据，实时指标又得另起一套逻辑，口径很难对齐。Flink SQL 的 Join 能力天然适合这种场景——既能做流与流的关联，也能做流与维表的关联，配合 Flink 自身的状态管理和精确一次语义，可以在一套引擎里完成所有打宽逻辑。

当然，方案选型只是第一步。真正把 Flink SQL 的 Join 用明白，要对不同 Join 的适用场景、状态原理、参数调优、问题排查都心里有数。下面的内容我会按照实际项目推进的顺序来写：先讲设计思路和 Join 选型，再拆解核心原理和实操细节，然后给出一个可复现的完整案例实现，最后整理我在实战中踩过的坑和排查方法。

1. 整体设计与 Join 方案选型

1.1 需求场景拆解：为什么需要打宽，以及打宽要解决什么问题

这个案例的业务口径其实不复杂：每产生一笔订单，我们需要知道这笔订单是哪个用户下的、用户的基础属性是什么样的、下单的商品是什么品类、价格是多少、以及这笔订单最终有没有完成支付、支付金额是多少。这些信息原本分散在几张不同的表里——订单表只有用户ID和商品ID这种关联键，用户维表才有姓名、等级、城市等信息，商品维表才有品类、类目、卖家信息，支付流水则是另一路独立的实时数据流。

如果只在离线环境做这件事，用一个标准 SQL 的 JOIN 就能搞定，跑批任务每天凌晨执行一次，把全量数据关联好写进宽表就行。但在实时场景下，问题就变复杂了：数据是源源不断到达的，用户信息可能在订单产生之后发生了变化，商品维表的数据不是静态快照而是会更新的，支付流水可能比订单数据晚到达几秒甚至几分钟，而且还存在订单一直没有支付成功的情况——这些都需要在流处理里特殊处理。

“打宽”这个词，本质上是在流式处理中把多张窄表（字段少、主题单一的表）通过关联键组合成一张宽表（字段多、信息完整）。它要解决的核心问题有三个：第一，关联键无法对齐时如何保证数据不丢；第二，参与关联的数据不断更新时如何保证结果准确；第三，延迟到达的数据如何处理才能不影响实时性。这些问题如果只靠简单地写个 JOIN 语句，是不可能真正解决的。

1.2 Flink SQL Join 的分类与选型逻辑

Flink SQL 的 Join 大致可以分为三类：Regular Join、Interval Join、Lookup Join（维表 Join）。这三类 Join 各有各的适用场景，在同一个打宽需求里甚至会组合使用。我先把它们放在一张表格里做个对比，再解释我在这个案例里为什么这样选。

Regular Join 是最通用的 Join，它把所有到达的数据都放进状态里，然后用 SQL 语义做等值关联。它的优点是语义最标准，任何一条左流数据都能关联到任意时刻到达的右流数据；缺点是状态会无限增长，必须依赖 TTL 机制来控制。

Interval Join 是 Regular Join 的一个特化版本，它要求两条流的关联数据必须在指定的时间间隔内到达。在这个案例里，支付流水通常会在订单创建后的几分钟内产生，所以订单流和支付流水流用 Interval Join 是最合适的——既能满足关联需求，又不会让状态无限膨胀。

Lookup Join 是维度表关联的标准方案。用户维表和商品维表是存储在 MySQL 里的，数据量不大但更新频率不高，不需要把全量维表加载到流处理状态里，而是在每一条订单数据到达时，实时去查询一次维表，把需要的字段拼接进来。这种方式的延迟低、状态开销小，而且维表数据在外部存储中更新后，新到的订单就能查到最新值。

1.3 我为什么没用单一 Join 一把梭

有些同学可能会问：这三类关联能不能全部用 Regular Join 来做？理论上可以，订单流关联用户维表流、商品维表流、支付流水流都可以用 Regular Join 实现。但这样做有两个问题。

第一，用户维表和商品维表的数据是更新流（Changelog Stream），不是简单的追加流。如果用 Regular Join 把订单流和维表流做关联，一旦维表数据发生变更，Flink 会基于变更前后值产生 UPDATE_BEFORE 和 UPDATE_AFTER 事件，导致下游收到回撤数据。在订单宽表场景下，这意味着一条订单数据可能会因为用户维表里的手机号或地址变更而产生多条更新记录。对于下游报表系统来说，这种频繁的更新会带来不小的消费压力。

第二，状态开销不可控。厂家说的维表可能只有几百 MB，但订单流一天可能产生几千万条数据，如果所有维表变更历史都堆在状态里做 Join，状态后端压力会非常大，而且这种关联方式根本用不上维表的高效索引能力，纯属用牛刀杀鸡。

所以我的方案是把三类 Join 组合起来：订单流关联支付流水流用 Interval Join，订单流关联用户和商品维表用 Lookup Join。这样既保住了实时性，又把状态开销控制在合理范围内。

2. Flink SQL Join 的核心机制与原理解析

2.1 Regular Join 是“记忆型”关联

要理解 Flink SQL 里各种 Join 的行为差异，得先理解它底层是怎么实现的。Regular Join 的实现逻辑可以简单概括为：把左右两条流的数据都存到 Keyed State 里，然后以 JOIN 条件中的等值 key 为标准，把同一 key 下的数据进行交叉匹配，匹配到的结果输出，匹配不到的也先留着，等对方数据到了再做匹配。

这个过程很像是在内存里维护了两张无限增长的“临时表”。凡是进入过 Flink 的数据都不会被轻易丢弃，除非达到了状态 TTL 过期时间。正因为这样，Regular Join 才能做到“任意时刻到达的数据都能关联上”。但代价就是状态膨胀——如果业务数据量很大，或者状态 TTL 设置得很大，TaskManager 的堆内存和 RocksDB 的磁盘空间都可能成为瓶颈。

在打宽案例里，我其实没有用 Regular Join，但我把它放在前面说，是因为它是理解其他 Join 的基础。你只有明白了 Regular Join 是怎么工作的，才能理解 Interval Join 是怎么在它的基础上做优化的，也才能理解为什么维表关联不建议用 Regular Join。

2.2 Interval Join：用时间边界给状态“上锁”

Interval Join 在语法上看起来跟 Regular Join 差不多，但它多了一个关键条件：时间边界。通常我们会在 ON 条件里加上类似o.order_time BETWEEN p.pay_time AND p.pay_time + INTERVAL '5' MINUTE这样的时间过滤条件。这个时间边界不是随便写的，它起到了两重作用。

第一重作用是裁剪数据。当一条支付数据到达时，Flink 只需要在订单状态里去查找那些时间在支付时间前后指定范围内的订单进行匹配，不需要全量扫描。反过来，当一条订单数据到达时，它也只会在支付流状态里搜索指定时间窗口内的支付记录。这样一来，参与 Join 的状态数据就被限制在了一个时间滑窗内，不会无限增长。

第二重作用是清理状态。Flink 可以基于时间边界推断出哪些状态数据已经不可能再被匹配了，从而安全地清除它们。比如我们设定了订单时间和支付时间最多相差 5 分钟，那么一条超过当前 watermark 5 分钟以上的订单数据，就不再需要保留在状态里了。

在这个案例里，订单流和支付流水流就用的是 Interval Join。业务上支付通常会发生在订单创建后的 3 分钟内，我留了点余量，设置了 30 分钟的上界。这个窗口不能设得太小，否则晚到的支付流水会关联不上；但也不能设得太大，否则状态数据量会变多，而且延迟数据补偿的实时性会受影响。

2.3 Lookup Join：维表关联的正确姿势

Lookup Join 是 Flink 1.9 开始引入的功能，它解决的是流式数据关联存储在外部系统中的维表数据时的性能问题。在我们这个场景里，用户维表和商品维表都存在 MySQL 中，如果每来一条订单数据都直接去查一次 MySQL，在高吞吐下数据库必然扛不住。Flink 的 Lookup Join 内置了缓存机制，可以显著降低对维表的访问频率。

Lookup Join 的核心机制可以这样理解：当一条流数据到达时，Flink 会提取出 JOIN 条件中的维表关联键（比如 user_id），然后去查询维表数据源。Flink 会首先检查本地缓存中是否有这个 key 的数据，如果有就直接用，如果没有才去查询外部系统，并把结果放入缓存。

这里的缓存策略有两个关键参数：lookup.cache.max-rows和lookup.cache.ttl。它们分别控制缓存的最大行数和缓存数据的存活时间。在实际项目中，TTL 的设置需要结合维表数据的更新频率来定。如果维表一天只更新一次，那缓存 TTL 可以设置得长一些，比如 1 小时甚至更长；如果维表是分钟级或秒级更新的，那缓存 TTL 就要相应缩短，否则查询到的是过期数据。

我在这个案例中把lookup.cache.ttl设置为了 30 秒。当时业务方反馈用户城市和商品类目偶尔会有变更，但不是高频更新，30 秒的 TTL 可以在数据新鲜度和数据库压力之间取得一个比较合理的平衡。

2.4 时间属性与 Watermark 在 Join 中的角色

Flink SQL 的流式处理离不开时间概念。针对 Interval Join 来说，时间属性尤其关键——必须在 SQL 中为参与 Join 的流明确定义时间字段，并且设置正确的 watermark 策略。

我在这个案例里统一使用了处理时间（Processing Time），原因是支付流水流的源头系统没有提供可靠的业务时间字段，只有到达 Kafka 的时间戳。如果你在真实的业务里能拿到可靠的 event time，那建议优先用 event time，因为它能正确处理乱序和延迟问题。但如果源数据本身的时间字段不可靠，硬用 event time 反而会引入偏差，不如老老实实基于处理时间做关联。

需要注意的是，如果选择了 processing time，那么在数据乱序严重或者某个上游任务出现故障重启的情况下，可能会产生关联不上的情况。这个项目的下游对准确性要求没那么苛刻，丢失个别延迟过大的数据是可以接受的，所以 processing time 的方案可行。如果你的业务场景对准确性要求极高，那就要设计延迟数据补偿机制了——这个我在后面问题排查部分会详细讲。

3. 完整案例实操：表的创建与 Join 打宽实现

3.1 环境与数据源准备

我用的是 Flink 1.17 版本，部署模式是 Flink SQL Client 配合独立集群。所有源表都定义在 Kafka，维表在 MySQL，结果表也写到 Kafka，下游由另一个任务负责消费宽表数据写入 ClickHouse。

先看源表定义。订单表的 DDL 如下：

CREATE TABLE orders ( order_id BIGINT, user_id BIGINT, product_id BIGINT, order_amount DECIMAL(10, 2), order_time TIMESTAMP(3), proctime AS PROCTIME(), WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'ods_orders', 'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092', 'properties.group.id' = 'flink-sql-join-group', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json', 'json.fail-on-missing-field' = 'false' );

这里我单独说一下proctime AS PROCTIME()这一行。它在表上定义了一个 processing time 属性，但这个字段不会从 Kafka 消息里读取，而是 Flink 自动生成的机器时间。Lookup Join 的语法要求维表关联时，流表必须有一个 processing time 字段，否则会直接报错。这个细节很多人第一次写的时候都会漏掉。

接着是支付流水表：

CREATE TABLE payments ( pay_id BIGINT, order_id BIGINT, pay_amount DECIMAL(10, 2), pay_time TIMESTAMP(3), pay_status INT, proctime AS PROCTIME() ) WITH ( 'connector' = 'kafka', 'topic' = 'ods_payments', 'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092', 'properties.group.id' = 'flink-sql-join-group', 'scan.startup.mode' = 'earliest-offset', 'format' = 'json', 'json.fail-on-missing-field' = 'false' );

然后是用户维表和商品维表。这两张表都存储在 MySQL，使用 JDBC connector 读取：

CREATE TABLE dim_users ( user_id BIGINT, user_name STRING, user_level STRING, city STRING, phone STRING, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://mysql-host:3306/dim_db', 'table-name' = 'dim_user_info', 'username' = 'flink_user', 'password' = 'xxxxxx', 'lookup.cache.max-rows' = '10000', 'lookup.cache.ttl' = '30s' ); CREATE TABLE dim_products ( product_id BIGINT, product_name STRING, category STRING, price DECIMAL(10, 2), seller STRING, PRIMARY KEY (product_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://mysql-host:3306/dim_db', 'table-name' = 'dim_product_info', 'username' = 'flink_user', 'password' = 'xxxxxx', 'lookup.cache.max-rows' = '5000', 'lookup.cache.ttl' = '30s' );

3.2 打宽 SQL 的实现与解读

数据源和维表都定义好之后，核心的打宽逻辑就是一个 INSERT INTO 语句。我完整贴出来：

INSERT INTO dwd_order_wide SELECT o.order_id, o.user_id, o.product_id, o.order_amount, o.order_time, u.user_name, u.user_level, u.city, p.product_name, p.category, p.seller, pay.pay_id, pay.pay_amount, pay.pay_time, pay.pay_status FROM orders o LEFT JOIN dim_users u ON o.user_id = u.user_id LEFT JOIN dim_products p ON o.product_id = p.product_id LEFT JOIN payments pay ON o.order_id = pay.order_id AND pay.pay_time BETWEEN o.order_time AND o.order_time + INTERVAL '30' MINUTE;

整个 SQL 看起来简洁，但里头每个 JOIN 的行为我都详细说下。

第一个LEFT JOIN dim_users u和第二个LEFT JOIN dim_products p是 Lookup Join。它们的语义是：当订单流里每来一条数据，Flink 将根据user_id去 MySQL 维表里查用户信息，按照product_id去查商品信息。由于前面建流表时定义了proctime，Flink 才能识别这些是 Lookup Join。

第三个LEFT JOIN payments pay是 Interval Join。跟前面两个不同，它关联的是两条流。条件里的pay.pay_time BETWEEN o.order_time AND o.order_time + INTERVAL '30' MINUTE是时间边界条件，它保证了支付流水只会在订单发生后的 30 分钟内被关联到。

这里有个很容易踩的坑：Interval Join 要求左右两边的流都必须有 watermark。我在orders表上定义了 watermark，但在payments表上只定义了 proctime 没有 watermark。这种情况下 Flink 会默认使用 processing time 作为 watermark 的来源吗？不会。实际上如果 Interval Join 的参与表没有定义合理的 watermark，任务在编译阶段就过不去。我在实际调试时发现这里报错后，又在payments表上补了一个 watermark 定义。为了统一，我在这里给payments也加上WATERMARK FOR pay_time AS pay_time - INTERVAL '5' SECOND。

如果使用的是 Flink 1.17 及以上版本，在流表上可以通过WATERMARK FOR来声明。注意pay_time字段必须能转换成 TIMESTAMP(3) 类型，JSON 里需要是字符串形式的时间，比如"2024-05-20 10:30:00"。

3.3 结果表定义与提交运行

宽表结果我们写入 Kafka，供下游消费：

CREATE TABLE dwd_order_wide ( order_id BIGINT, user_id BIGINT, product_id BIGINT, order_amount DECIMAL(10, 2), order_time TIMESTAMP(3), user_name STRING, user_level STRING, city STRING, product_name STRING, category STRING, seller STRING, pay_id BIGINT, pay_amount DECIMAL(10, 2), pay_time TIMESTAMP(3), pay_status INT, PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'kafka', 'topic' = 'dwd_order_wide', 'properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092', 'properties.group.id' = 'flink-sql-join-sink', 'format' = 'upsert-kafka', 'upsert-kafka.properties.bootstrap.servers' = 'kafka-1:9092,kafka-2:9092', 'key.format' = 'json', 'value.format' = 'json' );

这里我使用了upsert-kafka作为结果表的 format。为什么不用普通的 json？因为宽表数据中order_id是主键，同一笔订单可能会因为支付流数据的到达产生一条新记录，用 upsert-kafka 可以保证下游按主键去重，只保留最新状态。如果直接用 json 格式，Kafka 里会存在多条同主键的重复数据，下游再处理就需要额外的去重逻辑。

写完这些 DDL 和 INSERT 语句，在 Flink SQL Client 里依次执行即可。启动之后，可以通过 Flink UI 观察作业的运行情况，重点看几个指标：numRecordsInPerSecond、numRecordsOutPerSecond、当前状态大小和内存使用情况。

3.4 参数配置的经验值

我用 Flink 的 Official 参数做了一些针对 Join 场景的调优，下面这几个参数值得重点关注。

其中table.exec.state.ttl对于 Interval Join 来说，可以设置得比时间边界大一些。比如时间边界是 30 分钟，状态 TTL 至少得是 1 小时，给延迟数据留出缓冲。如果状态 TTL 设置得比时间边界还小，有些还没被关联到的数据就可能被提前清理，导致数据丢失。

我还调整了上下游并行度的配置。读取 Kafka 的源算子并行度设置成跟 Kafka 分区数一致，维表 Lookup 的处理算子并行度适当加大，因为每个并行子任务都会有自己的缓存，缓存命中率会更高。

4. 常见问题与排查技巧实录

4.1 维表关联不上：数据到了但查不到

这个问题的现象是宽表里维表字段全是 null。排查思路分三步：先看维表里是否真的有对应数据，再看 Lookup Join 的关联键类型是否一致，最后看缓存和连接池是否正常。

我在实际项目中遇过一次比较隐蔽的问题：订单表里的user_id在 JSON 中是字符串类型的"1001"，而维表 MySQL 的user_id是 BIGINT 类型。Flink SQL 在做等值 Join 时，两边的数据类型不一致会导致关联不到任何数据，而且不会报错。因为 SQL 在编译阶段不会做强制的类型转换，"1001"和1001L在比较时直接判定为不相等。

解决方法是统一源表 DDL 中的字段类型。具体做法是，在定义orders表时把user_id声明为 BIGINT，然后在 JSON 解析时让 Flink 自动做类型转换（前提是 JSON 中的字符串值可以被合法的解析成 BIGINT）。如果源数据格式不好控制，可以用CAST(user_id AS BIGINT)显式转换，但注意要放在子查询里做，避免在 Join 条件中写 CAST 导致无法下推。

4.2 Interval Join 没有关联上支付数据

这个问题的典型表现是订单宽表已经输出了，但pay_id和pay_amount还是 null，支付数据明明在 Kafka 里也有。我排查后发现了两个原因。

第一个原因是我最开始建payments表时没有定义 watermark。在 Flink 的 Interval Join 实现中，没有合理 watermark 会导致定时器无法触发状态清理和关联判断。加上 watermark 之后问题就消失了。

第二个原因是时间边界设得太小。最开始我设的是 5 分钟，但业务方经常因为支付系统处理耗时较长，实际支付时间和订单时间的间隔超过 5 分钟。把这些延迟较大的支付数据过滤掉了。后来把边界调整到 30 分钟，关联率从 93% 左右提升到了 99.7%。

4.3 回撤流导致下游数据量暴增

这个问题在 Regular Join 场景中特别常见。订单流和支付流水流如果用 Regular Join 来做，当支付流水流产生数据后，Flink 会先输出一条基于未关联状态的订单数据，等支付流水到达后，再输出一条更新后的订单数据。这就意味着一条最终记录带了两次输出——一次是回撤，一次是新值。

如果没有把回撤逻辑处理好，下游直接消费 Kafka 的话，会出现大量重复数据。我在这个案例里用到两种方式规避：

第一是优先使用 Interval Join，因为它不会产生回撤流，输出的是 Append 流。第二是如果必须用 Regular Join，结果表必须用 upsert-kafka 这类支持主键去重的存储，下游再消费时就能拿到最终一致的宽表数据。

重要提示：在实时数仓里，回撤流不是 bug，而是 Flink 为了保证精确一次语义产生的正常行为。使用方必须理解：-U表示撤回上一条，+U表示更新为当前值。如果你向不支持更新语义的存储写入数据，需要在下游显式做聚合或去重处理。

4.4 维表缓存导致的数据新鲜度问题

业务方最开始反馈过一个现象：用户在 App 里修改了个人资料后，实时宽表里的用户信息要很久之后才更新。我查看任务发现lookup.cache.ttl设置了 1 小时，这就是导致延迟的原因。

这个问题的本质是 Lookup Join 的缓存机制。为了降低外部系统压力，Flink 在缓存有效期内不会重新查询维表。修改 TTL 可以直接解决，但需要评估维表所在数据库的承受能力。如果维表一天变更量不大，但业务要求变更后能尽快反映到实时链路里，可以用两种方案：一是把 TTL 缩短（比如 10 秒），二是改造成 CDC 方式，把维表的变更事件实时同步到 Kafka，再用流流 Join 的方式关联。第二种方案复杂度高，但对数据新鲜度要求极高的场景值得考虑。

4.5 状态太大，内存不够用

运行一段时间后，我观察到有一个 TaskManager 的堆内存使用率持续在 90% 以上，最终触发了 GC 频繁和作业重启。排查后定位到是table.exec.state.ttl设置过大，加上订单流的数据量大，导致 Keyed State 里积压了太多未关联数据。

我的处理方式是：把状态后端从 HashMap 切换到 RocksDB，同时在 SQL 里通过合理使用 Interval Join 缩小状态数据量。RocksDB 状态后端把状态存在本地磁盘，内存只保留部分热数据，能有效缓解堆内存压力。另外也为每个 Join 节点单独设置了 TTL，确保不会被一个超大状态拖垮整个作业。

4.6 常见问题速查表

4.7 关于数据倾斜的补充排查

在流式 Join 里，数据倾斜往往表现为某个 key 的数据量特别大，比如某个爆款商品的订单特别多，或者某个大客户的下单频率特别高。这会导致处理这个 key 的算子负载远高于其他算子，形成长尾。

Flink SQL 是自动按关联 key 分区的，流上的倾斜无法像离线那样通过加盐打散来彻底解决，因为那样会破坏 Join 的正确性。我通常的做法是：如果确认某个 key 占比过高，且对实时性要求可以容忍，就在上游把这一部分数据单独拆出来用一个低并行度作业处理；如果无法拆分，就通过调大该算子并行度来分摊负载。这个项目里商品维度会出现爆款效应，我把 Lookup Join 相关节点的并行度调到了 32，同时配合维表缓存，整体表现稳定。

5. 实战总结与后续扩展

这个案例整体从需求评审到上线运行花了大概两周时间。第一周主要花在上下游数据源的梳理和 DDL 调试上，第二周集中处理了 Join 关联率、状态大小、维表缓存这几个核心问题。上线后运行了一个月，宽表数据的关联成功率稳定在 99.7% 以上，下游报表的指标口径和离线宽表能对上 98% 左右——剩下的 2% 差异主要来自实时数据里尚未完成支付的订单，这是业务逻辑本身的差异，不是计算错误。

最后分享两个我在实际操作中体会比较深的小技巧。

第一个是关于状态 TTL 的设置思路。很多人把状态 TTL 当成一个全局配置拍脑袋就定了，其实 TTL 的选择应该完全跟着关联键的数据生命周期走。比如订单和支付的间隔最多 30 分钟，那 TTL 设置为 2 小时就绰绰有余；如果某个 Join 是流和维表的关联，维表数据本身不存在“过期”的概念，那 TTL 可以设置得长一些。分别设置不同 Join 的 TTL，比全局统一设置高效得多。

第二个是维表缓存的命中率监控。在 Flink UI 上可以查看 Lookup Join 算子的吞吐和延迟，如果发现维表查询吞吐高、延迟大，大概率是缓存 TTL 设短了或者并行度不够。我之前遇到过 MySQL 维表被频繁打成热点的情况，就是因为缓存设置的 max-rows 太小，热门 key 的缓存很快被挤掉。后来把lookup.cache.max-rows和并行度都加了一倍，问题立刻缓解。

这个方案后续还可以扩展的方向就是前面提到的维表 CDC 化。如果业务对维表数据实时性要求变高了，可以把 MySQL 的 binlog 通过 CDC 工具同步到 Kafka，维表变成一张实时的维度流，再用 Regular Join 或者 Interval Join 跟订单流做关联。这样能够达到近实时的维表更新效果。不过这也对状态管理和下游存储提出了更高的要求，到时候需要重新评估链路成本。