Flink在国内实时计算领域已经是事实标准了但真正把机器学习模型塞进Flink作业里做实时推理的团队其实并没有想象中那么多。大部分人的实践还停留在“用Flink做实时数仓把特征加工好落到离线表再在离线训练里用”的阶段。这当然没问题但如果你要做的场景是实时风控、实时推荐、实时异常检测那模型上线的路径就得重新考虑了。这篇东西并不是想给你一个放之四海皆准的银弹方案而是把我自己在实际项目中把Flink和AI模型集成起来、跑通实时推理的整个思路和踩坑经历梳理一遍。你会看到为什么我最终选择通过自定义UDF做嵌入式推理也会看到JDBC连接器、状态管理、背压这些绕不过去的坑是怎么处理的。1. 先把架构想清楚Flink集成机器学习的两条主流路径1.1 为什么非要用Flink来做模型推理很多同学第一个疑问是模型推理不是直接写个Python服务、暴露一个HTTP接口就完了吗为什么非要绕一圈塞进Flink里如果你的业务场景只是“拿到一个请求调用模型返回结果”那独立推理服务完全够用。但如果你要的是“每秒钟几万条流式日志进来每条都需要打上模型分数而且分数要跟窗口聚合结果、历史状态一起参与计算”独立服务的方案就会很难受。我之前做过一个实时反欺诈项目数据源是Kafka里的用户行为日志每秒峰值大概在两万条左右。每条日志需要先做特征拼接把用户近5分钟的点击序列、设备指纹、IP风险等级等二十几个特征拼出来然后送进一个XGBoost模型打分分数超过阈值的直接触发风控规则。如果走HTTP调用独立推理服务网络开销、序列化开销、并发连接数瓶颈会立刻变成系统瓶颈而且Flink作业的Checkpoint也会因为外部依赖的不稳定而频繁超时。把模型直接嵌进Flink算子内部让推理发生在同一份内存数据流过的地方就不需要跨进程通信单条数据的端到端延迟可以控制在几十毫秒以内。同时还能借助Flink的窗口、状态、Watermark机制让特征拼接和模型打分在同一个作业里闭环完成。这是Flink做实时推理最核心的价值它把特征工程、模型推理、规则引擎和下游Sink整合成了一个一致性的流式管道。1.2 两条主流技术路径对比目前业内做Flink和AI集成的方案抽象来看其实只有两条路。第一条是嵌入式推理我把训练好的模型序列化文件比如PMML、ONNX、或者Java直接加载的字节码扔到Flink的分布式缓存里在UDF或者ProcessFunction里完成模型加载和预测。这种方法的好处是低延迟、没有外部依赖、架构简单缺点是模型文件必须能被打进Flink的运行环境模型更新要走作业重启或者版本切换机制。第二条是外部推理服务Flink算子通过网络调用一个独立部署的推理引擎可以是TensorFlow Serving、ONNX Runtime Server、或者你自己写的Python Flask服务。好处是模型更新灵活训练团队可以独立迭代模型而不需要动Flink作业坏处就是前边说的延迟、稳定性和并发问题都得靠你自己做连接池、超时控制、降级策略来兜底。我做过的项目里两条路都走过。如果是模型文件比较小几十MB以内推理逻辑相对固定对端到端延迟要求极高我强烈建议用嵌入式推理。如果模型要频繁更新每天甚至每小时版本迭代或者模型结构复杂需要GPU资源池化那就老老实实把模型部署成独立服务Flink侧通过异步I/O配合连接池去调用同时要做好限流和服务降级的预案。1.3 我判断方案选型的四个关键条件拿实时反欺诈项目举例我当时判断要不要做嵌入式推理主要看四件事。模型大小是否可控。XGBoost的模型文件在30MB左右每个TaskManager加载一份内存占用可以接受。如果是一个PyTorch的Embedding模型动辄几个GB那Flink的TaskManager内存必然会被拖垮这种就适合走外部服务。特征拼接是否依赖Flink状态。如果特征需要用到窗口聚合、会话超时判断这些状态计算那嵌入式方案明显更顺因为特征计算和打分在同一个算子里状态不用跨系统传递。反过来特征全在外部特征存储里那Flink只是当一个纯粹的推理客户端外部服务反而更清晰。延迟指标是否极其苛刻。实时竞价的场景延迟要求P99在50ms以内外部服务的网络开销基本不可能稳定做到所以只能嵌入式。但像风控反欺诈这种P99在200ms就能接受的场景两者也有得商量。团队的维护边界。如果模型由算法组维护他们希望完全掌控模型发布流程那就别让模型跟Flink作业绑在一起。嵌入式推理的痛点在于模型更新往往要跟着Flink作业发版走运维协调成本很高这个必须得想清楚。2. 模型接入与特征工程最容易翻车的环节2.1 模型加载和分布式缓存的使用细节嵌入式推理第一步是把模型文件分发到所有TaskManager上。Flink官方提供的方式是分布式缓存作业提交时把模型文件注册进缓存各个算子第一次执行时可以顺着缓存路径读取模型文件。实际操作中我建议不要直接在RichFunction的open方法里做庞大模型的加载和解析因为open阶段是JobManager在启动所有并行子任务时统一调用的加载太慢会造成整个作业启动超时。更靠谱的做法是懒加载。第一次有数据到达算子时再去初始化模型用一个volatile或者AtomicReference持有着模型实例后续的数据直接复用已经加载好的对象。这样作业的恢复时间不会因为模型加载而拖长也算是在工程上避开了Flink启动超时的经验坑。Kryo序列化这块需要额外上心。模型文件本身不参与Flink的状态序列化这个很多人会误解。我的经验是把模型文件作为一个transient字段放在函数类里每次执行前检查一下是否需要重新加载否则默认Java序列化会尝试把这个对象序列化了一遍又一遍严重影响作业的整体吞吐。2.2 在线特征拼接窗口、状态和迟到数据真实场景里模型打分很少只依赖单条数据的原始字段基本都要靠聚合特征。比如我看一个用户当前的点击行为是否异常需要知道他最近5分钟点击了多少次、最近1小时的品类分布是什么、以及7天里有没有相似的行为序列。这些特征没法从单一事件里直接拿到必须用Flink的窗口和状态帮你攒着。这里有一个特别容易被忽略的细节窗口计算的任务和模型推理的任务到底要不要放在同一个算子里。我自己的实践经验是如果聚合特征是计算在一个窗口算子里的而模型推理放在下游另一个算子那么连接操作会引入不必要的数据Shuffle每条数据都要经过一次网络传输。反欺诈这种场景本身数据量大比较理想的做法是在ProcessFunction里手动管理状态用State作为滚动窗口每条数据到达时更新状态、计算特征、再直接送入模型打分。这样特征计算和推理发生在同一个算子上没有Shuffle状态访问也是本地性能会好不少。迟到数据是另一个坑。训练模型时标签和特征是对齐的但实时场景里特征可能因为乱序、迟到而不完整打分自然就不准。处理迟到数据我一般会给事件时间设置一个允许的延迟时间比如5秒超过这个窗口的数据如果还想参与特征聚合就需要判断当前特征里到底有多少信息是新补充的。实在拿不准的时候宁可让模型拿旧特征打分也千万不要因为特征缺失直接抛异常这种粗鲁的处理方式会导致下游告警刷屏。2.3 数据源的接入方案Kafka和MySQL CDC大多数实时推理作业的数据主路径是Kafka但在实操里我经常看到别人把特征辅助数据也硬编码进作业里这是非常不灵活的。正确的做法是把维表数据通过维表Join的方式挂进来Flink SQL的维表Join因为支持Lookup Cache在处理低频变更的维度数据时性能还是相当不错的。值得一提的是MySQL CDC源这个热搜词里出现的频率特别高现实中真的是一个高频场景。你要把MySQL的数据同步到ClickHouse里做后续分析或者同步到本地Cache供实时推理查特征维度用Flink CDC是最省力的方案。但CDC Source有一个天然的坑如果是全量加增量模式第一次启动时会做全表扫描这个阶段会把数据库的连接资源占得很满如果不加限制直接跑业务高峰期可能会把数据库拖挂。我自己处理这个问题时会在作业提交前先用一个低峰期时间段专门跑全量同步等全量完毕再切换成增量模式或者在SQL层面配合数据库的WHERE条件做分批扫描把一次性全表扫描量给拆散。3. 实操过程从完成训练的模型到实时推理作业3.1 环境与依赖计划的准备如果你要从零开始做一个Flink加AI的实时推理作业首先别急着写代码先把依赖和运行环境的版本对应关系理清楚。以下是我比较常用的一套组件组合你参考的时候注意对齐你自己的Flink版本Flink版本1.17或1.18Flink SQL和DataStream API都可用Python环境如果是走Python UDFFlink的PyFlink需要单独安装Python 3.7到3.10之间的版本模型格式XGBoost模型直接转成PMML或者转成ONNX两种我都试过ONNX对Python和Java的跨语言支持更好推理引擎ONNX Runtime的Java版本maven坐标是com.microsoft.onnxruntime:onnxruntime比较推荐的方式是先把模型文件放到Flink的分布式缓存里作业启动时再读取。从工程化角度来看模型文件建议上传到HDFS或者S3上然后在Flink命令行通过-yD之类的参数指定缓存路径。如果你用的是Flink SQL Gateway或者YARN应用模式记得给JobManager足够的内存来缓存这些模型文件的下载过程。3.2 核心代码基于ONNX的嵌入式推理UDF以一个XGBoost转ONNX的实时评分模型为例我给你看一个完整的Java UDF实现思路。这里追求的是能跑通很多细节比如指标记录、日志输出你都可以根据自己场景加。import org.apache.flink.api.common.functions.RichFunction; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.util.Collector; import ai.onnxruntime.OnnxTensor; import ai.onnxruntime.OnnxValue; import ai.onnxruntime.OrtEnvironment; import ai.onnxruntime.OrtSession; import org.apache.flink.api.java.utils.ParameterTool; public class RiskScoreFunction extends ProcessFunctionFeatureEvent, ScoredEvent { private transient OrtEnvironment env; private transient OrtSession session; private transient byte[] modelData; Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 从分布式缓存读取模型文件 // 这里假设模型已经注册为缓存文件名称是 risk_model.onnx java.io.File modelFile getRuntimeContext().getDistributedCache().getFile(risk_model.onnx); byte[] data java.nio.file.Files.readAllBytes(modelFile.toPath()); this.modelData data; } private void ensureModelLoaded() throws Exception { if (env null) { env OrtEnvironment.getEnvironment(); } if (session null) { session env.createSession(modelData); } } Override public void processElement(FeatureEvent value, Context ctx, CollectorScoredEvent out) throws Exception { ensureModelLoaded(); long startTime System.currentTimeMillis(); float[] tensorData extractFeatures(value); OnnxTensor tensor OnnxTensor.createTensor(env, tensorData); OrtSession.Result result session.run(java.util.Collections.singletonMap(input, tensor)); float score (float) ((OnnxValue) result.get(score).get()).getValue(); out.collect(new ScoredEvent(value, score)); // 打印推理耗时方便定位性能问题 if (System.currentTimeMillis() - startTime 50) { System.out.println([warn] infer slow: (System.currentTimeMillis() - startTime) ms); } } }这段代码的核心思想是懒加载模型open阶段只读取文件字节真正的模型初始化放在第一条数据到来时做。实际运行中session创建非常耗时大概几百毫秒所以同一个TaskManager上的所有数据复用同一个session实例这是性能的关键。注意这里的processElement里可以加自己的特征拼接逻辑也可以在进入这个算子之前用Flink SQL完成特征聚合然后把字段传入。我个人倾向这条是经验之谈把特征拼接做到SQL侧让Flink SQL去处理窗口聚合、维表Join最终用一条简单的SELECT语句把拼接好的特征行交给这个UDF。理由是SQL的可维护性强而且Flink SQL的优化器会自动处理部分算子下推比手动写ProcessFunction处理窗口要省心很多。3.3 用Flink SQL结合UDF实现完整推理流程在实际生产里直接用纯DataStream API写整个作业的可维护性很差。我现在的做法是大部分时候用Flink SQL做数据接入、清洗、窗口聚合然后通过自定义UDF把推理逻辑嵌进SQL链路里。下面是一个比较典型的作业结构-- 1. Kafka源表 CREATE TABLE user_behavior ( user_id BIGINT, item_id BIGINT, behavior STRING, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL 5 SECOND ) WITH ( connector kafka, topic user_behavior, properties.bootstrap.servers kafka-1:9092,kafka-2:9092, format json ); -- 2. MySQL维表用于补充用户风险等级等维度 CREATE TABLE user_dim ( user_id BIGINT PRIMARY KEY, risk_level INT, device_fingerprint STRING ) WITH ( connector jdbc, url jdbc:mysql://mysql-host:3306/risk_db, table-name user_dim, username flink_user, password secret ); -- 3. 推理结果表输出到Kafka供下游规则引擎消费 CREATE TABLE risk_score_result ( user_id BIGINT, item_id BIGINT, behavior STRING, ts TIMESTAMP(3), risk_level INT, score DOUBLE ) WITH ( connector kafka, topic risk_score_result, format json ); -- 4. 核心推理逻辑 INSERT INTO risk_score_result SELECT t.user_id, t.item_id, t.behavior, t.ts, d.risk_level, RiskScorer(t.user_id, t.item_id, t.behavior, d.risk_level, d.device_fingerprint) AS score FROM ( SELECT * FROM user_behavior ) t LEFT JOIN user_dim FOR SYSTEM_TIME AS OF t.ts AS d ON t.user_id d.user_id;这里最关键的是RiskScorer这个UDF它接收拼好的特征字段内部做模型推理返回一个DOUBLE类型的分数。UDF在Flink中的注册方式很简单只要把实现类打成Jar包放进FLINK_HOME/lib然后在SQL客户端或作业代码里注册即可。用这个结构的好处是你在开发环境里可以直接查Kafka的原始数据、观察中间聚合结果、甚至单独测试UDF而上线生产时只需要调整并行度和状态后端配置。3.4 从MySQL同步到ClickHouse这个高频延伸场景热搜词里关于MySQL同步到ClickHouse的出现频率相当高这也跟Flink的一个重要应用方向相关。实时推理的特征辅助数据很多时候不是存在MySQL里的而是存在ClickHouse里做后端分析用的。但特征数据本身需要很低延迟的读取所以很多团队的架构会做两层一层是MySQL作为业务主库一层是ClickHouse作为分析库和部分特征存储。Flink CDC做这两者之间的同步标准做法是先定义MySQL的CDC源表再定义ClickHouse的Sink表中间做个简单的数据清洗转换。需要注意ClickHouse的JDBC连接器在同步高频小数据量写入时连接池复用非常重要。默认情况下每次写入都新建连接QPS稍微一起来就会出现Too many connections错误。所以我一直建议写ClickHouse的Sink时用批量写入加每隔N条或每隔N毫秒触发一次flush的方式。CREATE TABLE user_dim_cdc ( user_id BIGINT, risk_level INT, device_fingerprint STRING, PRIMARY KEY (user_id) NOT ENFORCED ) WITH ( connector mysql-cdc, hostname mysql-host, port 3306, username flink_cdc, password secret, database-name risk_db, table-name user_dim, scan.incremental.snapshot.chunk.size 10000 ); CREATE TABLE clickhouse_sink ( user_id BIGINT, risk_level INT, device_fingerprint STRING ) WITH ( connector clickhouse, url clickhouse://ch-host:8123, table-name user_dim_sync, sink.batch-size 500, sink.flush-interval 2000 ); INSERT INTO clickhouse_sink SELECT user_id, risk_level, device_fingerprint FROM user_dim_cdc;这套方案跑起来以后运维很省心CDC会监听MySQL的Binlog自动捕获变更ClickHouse里的数据也基本能做到秒级延迟。值得注意的是使用CDC时如果修改了表结构Flink作业不一定能平滑适配所以生产环境里的表和表字段变更一定要提前在测试环境验证一遍。4. 常见问题与排查技巧实录4.1 JDBC连接器异常的排查JDBC连接器异常几乎每个做过Flink联表或写库的同学都会碰到我最早也被查得满头包。最典型的报错是Cannot connect to MySQL server或者Communications link failure不是因为你的密码写错了而是连接数超过了MySQL的上限。Flink的JDBC源表默认会为每个并行子任务创建一个连接如果你作业的并行度是20意味着同一时刻最多可能有20个连接在轮询表再加上一些刷新、检查任务MySQL的max_connections如果只开了一百多很容易被打爆。解决思路分两步。第一步是给JDBC连接器的URL加上连接池参数比如useSSLfalserewriteBatchedStatementstrueuseServerPrepStmtstruecachePrepStmtstrue。第二步是调小JDBC Source的轮询间隔或者把表改成增量Lookup Cache模式避免每条数据都去数据库查询。如果你是要同步大批量数据建议直接用Flink CDC而不是JDBC扫描CDC做的是Binlog监听连接数占用非常少。还有一个隐蔽问题JDBC连接器的时区设置。Flink读取MySQL时间戳字段时如果JVM默认时区和MySQL的时区不一致会出现时间偏差8小时的问题。这个在SQL的CREATE TABLE里显式设置scan.fetch-size或者直接修改连接串的serverTimezoneAsia/Shanghai就能解决。4.2 模型推理慢导致背压该怎么定位和解决嵌入式推理最担心的是模型推理耗时不稳定一旦某个算子处理速度赶不上上游数据到达速度就会产生背压然后Checkpoint开始频繁超时整个作业进入恶性循环。如果你在Web UI的BackPressure页签看到某个算子显示HIGH优先确认是不是推理算子的峰值耗时太长。我自己总结了一个三步定位法第一步在推理算子的processElement里打印前1%的耗时日志看耗时集中在哪。大部分时候是特征数量变多导致Tensor维度变化或者模型自身计算量对某一批特征矩阵特别敏感。第二步看是不是模型加载速度导致的比如每秒第一次推理触发了session的初始化后续的推理反而很快。这种问题很容易通过一个定时预热作业提前把模型加载到内存让它在正式流量进来前已经初始化完毕。第三步看并行度是否足够。如果你用10个并行度处理每秒5万条输入但模型推理本身的吞吐上限只有每秒3000条那就是纯算力不足。此时要么调高并行度要么改模型格式比如将XGBoost的少数高开销特征做前置截断或者改用更轻量的规则模型过滤掉明显不会触发的数据。我还试过一种比较取巧的办法如果打分模型比较大但业务允许的话可以在特征拼接阶段做一次轻量预筛把不可能触发阈值的那些数据直接打0分只让真正有风险的数据走完整模型。这种做法的收益很可观但也有风险就是预筛规则和模型效果必须高度一致否则误杀率会高到不可接受。4.3 模型版本更新与状态兼容模型更新是嵌入式推理最麻烦的事。更新模型文件很容易但Flink作业如果开着Checkpoint或者Savepoint状态里存的可能还是旧版本的模型相关的特征格式两者对不上就会出现反序列化失败。我的经验是如果只是模型参数更新、输入输出格式不变可以直接替换分布式缓存里的模型文件然后从最近一次Savepoint恢复作业即可不需要改代码。但如果输特征格式变了比如新增了一些特征字段那就要先调整特征拼接逻辑再做一次无状态重启让作业从最新状态开始跑。生产环境的模型版本管理我建议这样处理在推理算子内部维护一个模型版本号字段每处理一条数据就打一个版本标签下游分析和规则引擎以后也能清楚地知道这条记录是由哪个模型打的分数。不然模型回滚或者对比效果的时候数据口径一片混乱。4.4 关于GPU和模型推理资源的一些经验热搜词里提到L20显卡适合部署什么模型说明现在很多人都在关注GPU推理资源。如果推理服务的算力确实扛不住常规做法是上GPU。但Flink本身的TaskManager是CPU密集型的Streaming计算框架你在一个TaskManager进程里同时跑GC任务和GPU推理资源隔离并不好做。我的经验是不要把GPU推理直接塞进Flink算子内而是把它当作外部推理服务Flink通过异步I/O去调用这样GPU可以独立扩缩容。GPU推理服务的部署方式可以参考当前比较主流的推理引擎套件比如Triton这类不同的模型框架都能统一暴露成HTTP或gRPC接口。实操中异步I/O最容易被忽略的是并发数和队列容量。异步I/O如果并发数设置太大会对推理服务造成极大压力设置太小则发挥不了异步优势。我给的参考值连接数大约是推理服务最高QPS的两倍超时时间设置在100ms到200ms之间如果超时了宁可降级也不要把错误直接抛到Flink算子中。再补充一个小技巧突发流量时推理服务会排起长队Flink侧看到的背压其实是从外部服务传导过来的。此时与其不停扩容Flink并行度不如在推理服务前面加一层缓存或者把请求分级高优先级先处理低风险的、低优先级后处理否则扩容只会让外部服务更快地崩溃。我个人在实际操作中最大的体会是Flink接AI模型这件事真正难的从来不是“能不能调用模型”而是“如何保证作业在复杂的实时数据环境下稳定地持续产出正确结果”。所以做这类项目我建议你先花时间把数据管道打磨稳再考虑模型效果优化。模型不行可以换作业不稳一崩溃所有人的工作节奏就全乱了。最后分享一个小经验吧每次调试模型接入Flink的时候尽量先在本地用Flink MiniCluster跑一个小数据集模型文件也用本地路径调通之后再切到生产集群。这个习惯能帮你节省至少一半的排查时间。很多看起来像是模型推理的问题最后定位下来其实都是配置或者依赖冲突。先用最简单的方式跑通再逐步加复杂度这条原则永不过时。