Flink SQL 实时数仓打宽实战：Join 选型、原理与问题排查

Flink SQL 做流上数据打宽这几年已经算是实时数仓里的常规操作了。但我在不少交流群和线下分享里发现很多人对 Flink SQL Join 的理解还停留在“写个 JOIN 语法就能跑”的阶段真正遇到数据倾斜、状态膨胀、乱序补偿、延迟关联这些问题时往往手足无措。这篇文章我就结合自己实际做过的一个 FlinkSQL-Join 打宽案例把从需求拆解到方案落地、再到问题排查的完整过程都摊开来讲希望能给正在做实时数仓或正在琢磨 Flink SQL 的同学一些可以“抄作业”的参考。先说下这个案例的背景。当时业务方需要一个实时订单大宽表把订单主表、用户维表、商品维表、以及另一路支付流水流进行关联最终输出包含完整用户信息、商品信息、支付信息的宽表数据供下游实时报表和数据分析使用。整个链路如果用传统的 Lambda 架构去做离线 T1 的大宽表只能处理历史数据实时指标又得另起一套逻辑口径很难对齐。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 各有各的适用场景在同一个打宽需求里甚至会组合使用。我先把它们放在一张表格里做个对比再解释我在这个案例里为什么这样选。Join 类型数据形式是否支持更新状态存储适用场景Regular Join流与流支持会产生回撤流较大需保留所有参与数据双流对账、流与流任意关联Interval Join流与流支持限定时间范围相对较小只保留窗口内数据订单与支付流水关联、点击与曝光关联Lookup Join流与维表/外部存储维表更新可感知不存维表状态只存关联结果或被查流的必要信息流与 MySQL/HBase 维表实时关联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去查商品信息。由于前面建流表时定义了proctimeFlink 才能识别这些是 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.ttl1 h控制 Regular Join 和 Interval Join 的状态过期时间防止状态无限膨胀lookup.cache.ttl30 s维表缓存过期时间保证数据新鲜度的同时降低外部系统压力lookup.cache.max-rows10000维表缓存最大行数防止缓存占用过多内存execution.checkpointing.interval60 s开启 checkpoint保证故障恢复时数据不丢、不重execution.checkpointing.tolerable-failed-checkpoints3允许一定次数的 checkpoint 失败避免作业频繁重启其中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 常见问题速查表现象可能原因处理方案维表字段全为 null类型不一致或维表缺数据核对关联键类型、查询维表确认数据存在Interval Join 关联不上缺少 watermark、时间边界过小补 watermark、调大时间边界结果数据重复回撤流未处理使用 upsert-kafka 或下游按主键聚合维表数据更新不及时lookup.cache.ttl 过大按维表更新频率调整 TTL作业内存持续增长状态膨胀或 TTL 过大切换 RocksDB、调小状态 TTLJoin 算子并行度低默认并行度不匹配数据量单独设置 Join 算子并行度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 跟订单流做关联。这样能够达到近实时的维表更新效果。不过这也对状态管理和下游存储提出了更高的要求到时候需要重新评估链路成本。