云栖大会聊到 AI 进生产的时候，我坐在台下最大的感受是：今年已经没有多少人再去吹模型参数了。台上台下反复出现的焦点，是一句话——AI 应用进入生产，开始拼「实时数据智能」。这句话翻译成工程师听得懂的话就是：模型算法能力差不多拉平之后，谁的端到端链路短、谁的数据新鲜度够、谁能用实时数据持续修正模型效果，谁就能把 AI 真正变成业务指标。
这篇文章想跟你聊的，不是某个新模型或某个框架的用法，而是一条完整的实时数据智能链路：它到底是什么、为什么 AI 一进生产就绕不开它、技术架构怎么搭、从零开始怎么做，以及我在实战里踩过的坑。内容适合正要启动 AI 应用的算法工程师、数据工程师和架构师，也适合被业务方天天追问“为什么 AI 效果不稳定”的产品同学。
1. 为什么AI应用一进生产，就开始拼实时数据智能
1.1 模型能力变成入场券，数据新鲜度才是真正的护城河
先说实话，模型不是不重要，但在生产环境里它的权重正在下降。这个过程其实很好理解：开源模型和各家 API 把底层能力迅速拉平，大家都能用上差不多的基座模型，再靠提示词和微调拉开的差距，远没有 Demo 阶段看起来那么大。真正让应用效果分化的，是模型输入的数据，以及数据能不能跟上业务发生的节奏。
举个很直观的例子。你做的是一个智能助手，用户问“附近哪家店还有这款手机的现货”，模型本身的问答能力再强，如果不知道门店实时库存，就只能给一个语焉不详的通用回答。你再问他“现在下单什么时候能到”，模型不知道物流系统的实时状态，同样只能猜。这类问题不是靠把模型做大能解决的，它需要实时数据持续送到模型面前。用户能感知到的“智能”，本质上是“对当前情况掌握得有多清楚”。
到了生产环境，这个问题会被放大。模型用的特征如果是 T+1 离线算好的，那它做决策时看到的是昨天的用户行为、昨天的库存、昨天的价格。推荐系统可能给你推一件已经下架的商品，风控模型可能漏掉一笔刚发生的异常交易，营销系统可能把优惠券发给刚刚已经领过的用户。这些问题在离线评估里看不出来，因为离线评估用的是同样过期的数据，大家都“公平地错着”。但线上用户不跟你讲公平，他只会觉得这个 AI 不太聪明的样子。所以我说，模型能力已经变成了入场券，真正拉开差距的，是数据新鲜度，以及基于新鲜数据做决策的整套工程能力。
1.2 到底什么是「实时数据智能」
这个概念这两年被讲得很乱，有人把它等同于实时数仓，有人把它等同于在线学习。我个人的理解是：实时数据智能不是一个单点技术，而是一个闭环，至少包含四件事。
第一，实时数据接入和计算。业务事件从产生到被流式计算引擎处理好，中间经过消息总线、实时计算任务，最终变成可以直接使用的特征或者指标。第二，实时特征的存储和服务。算好的特征要能低延迟地被模型读取，通常放在 Redis、HBase 这类在线存储里，并且要有一套统一特征服务接口。第三，在线推理和决策。模型服务接收到请求后，把实时特征、用户画像、业务规则组合在一起，在毫秒到秒级的时间内给出决策。第四，反馈回流。决策产生了什么结果——用户点了没有、交易成了没有、设备故障了没有——这些结果要回到样本和特征链路里，形成持续的闭环。
你可以把这个闭环理解成一家店的经验丰富的店长。模型是店长的经验积累，他知道什么品类好卖、什么时间客流大；实时数据智能则是他一睁眼就能看到的现场情况——现在店里来了多少人、货架还剩几件、隔壁是不是在打折。经验都差不多的两个店长，谁能更及时地掌握现场，谁的经营决策就更准。注意，实时数据智能和实时 BI 不一样：BI 是算给人看的，核心是图表和报表；数据智能是算给机器用的，核心是直接驱动一个动作，比如推荐、拦截、定价、派单。前者人看完再决定，后者系统拿到结果马上执行，对延迟和稳定性的要求完全不在一个量级。
1.3 哪些场景最先被这件事改变
先看一张表，这是我见过的最典型的几类场景，它们的共性是：决策有时效性，数据有衰减性，超过一定时间再响应，结果就会大打折扣。
| 场景 | 决策内容 | 对实时性的要求 | 典型数据来源 |
|---|---|---|---|
| 个性化推荐与营销 | 下一屏展示什么、发不发券、发多大面额 | 秒级到分钟级 | 行为流、库存、价格、画像 |
| 风控与反欺诈 | 本次交易放行还是拦截、是否需要二次验证 | 毫秒到秒级 | 交易事件、设备指纹、关系网络 |
| 智能客服与助手 | 回答带实时信息的问题 | 秒级 | 订单状态、物流轨迹、库存 |
| 工业预测性维护 | 设备是否告警、是否派单 | 秒级到分钟级 | 传感器时序数据、工单系统 |
| 动态定价与供应链 | 价格调不调、补货补不补 | 分钟级 | 销量、竞对价格、库存水位 |
这些场景背后有一个共同规律：数据的价值随时间快速衰减。用户 3 分钟前的点击序列，比昨天的点击序列更能预测他现在的意图；一笔交易 100 毫秒内做风控判断，比事后 10 分钟再补查有意义得多。所以不是业务方突然变得“娇气”了，而是 AI 应用进生产之后，业务指标真实地长在数据新鲜度上。谁能把这条链路做扎实，谁就先拿到这个红利。
2. 实时数据智能的技术基座与设计要点
2.1 一条最小可用的架构长什么样
我见过很多团队一上来就想搞实时数据中台，结果半年过去了链路还没通。其实一条能跑通业务的最小架构并不复杂，核心就五层。
业务应用 / 设备埋点 ↓ 消息总线 (Kafka / Pulsar) ↓ 流式计算引擎 (Flink / Spark Structured Streaming) ↓ 特征存储与特征服务 (Redis / HBase / Feature Store) ↓ 在线推理与决策 (模型服务 + 规则兜底) ↓ 结果回流 (点击 / 成交 / 反馈日志 → 样本与模型迭代)消息总线负责削峰填谷，把业务系统跟计算层解耦。流式计算引擎负责做窗口聚合、维表关联、实时特征计算。特征存储负责把计算好的特征以低延迟方式提供给在线推理。模型服务负责把特征拼装成模型输入，打分，然后输出决策。最后，业务结果要原路回到样本系统，这是很多人忽略的一层，少了它，模型就没有办法针对线上分布的变化做迭代。
我还想强调一点：在线链路和离线链路从来不是两条独立的线。离线链路负责用全量历史数据训练模型、离线评估特征重要性；在线链路负责用实时数据做推理。两者之间必须共享同一份特征定义、同一套特征取值逻辑。如果离线特征和在线特征各算各的，后面一定会出现“训练指标很好、线上效果崩盘”的问题，这一点我在第三章和第四章都会反复提到。
2.2 离线在线一致性，比实时性本身更容易坑人
很多团队把精力全花在怎么把延迟压到 100 毫秒，结果上线后效果不行，一查发现是特征一致性问题。所谓一致性问题，就是离线训练时用的用户特征是“全量行为算出来的”，而在线服务时却只能拿到“最近一个时间窗口的行为”；离线时用了未来信息，在线时拿不到，模型自然会失效。
解决这个问题，最有效的办法是引入统一特征管理，也就是 Feature Store 的思路。每个特征必须有唯一的元数据定义，包含特征名称、计算逻辑、数据来源、窗口大小、更新频率、缺失值策略。离线训练和在线推理都从这同一份元数据生成代码，而不是各写各的。其次，离线训练要做 point-in-time join，也就是样本标签只能使用决策时刻之前已经存在的特征，绝对不能用决策时刻之后发生的信息。这个要求听起来基础，但实际操作中很容易在特征拼接时不小心把未来行为带进去。
还有一个容易被忽略的环节是特征回填。实时流任务上线之前，需要用历史日志把特征回填一遍，让离线训练样本能覆盖足够长的时间窗口。否则线上跑了三天才有足够的新样本，模型迭代周期就会被拖得很长。回填的逻辑必须和实时计算的逻辑完全一致，包括窗口划分方式、迟到数据处理方式。我建议直接复用同一套计算代码，离线回填和实时计算都从同一个任务模板生成，避免两套实现各自为政。
2.3 实时性怎么量化，怎么定目标
做实时数据智能，不能只有“快”这个模糊的感觉，要有明确的量化指标。我一般会盯三个东西：端到端延迟、数据新鲜度、SLA。
端到端延迟，指从业务事件发生到模型决策返回结果的时间差。这个指标要看百分位，重点看 p95 和 p99，因为平均值很容易被少数极快请求拉低，掩盖长尾问题。数据新鲜度，指事件进入系统到对应的特征可以被查询到的时间差。这个指标更贴近“数据有没有过期”的本质，比如你定的是 10 秒内特征更新，那么模型拿到的就不是几分钟前的快照。SLA 则是把前两者变成承诺，例如“端到端 p99 小于 500 毫秒，数据新鲜度小于 10 秒，月度可用性 99.9%”。
不同场景的目标差异极大。风控反欺诈可能要求 p99 在 300 毫秒以内，营销触达可以接受几十秒，工业预测性维护到分钟级也没有问题。千万不要一上来就追求毫秒级，实时性的投入产出比是严重非线性的：从一天压到一小时，可能只需要把批任务改成小时级调度；从一小时压到一分钟，需要上流处理；从一分钟压到一秒，需要做状态存储和在线特征服务；再往下压到毫秒级，就要开始抠序列化、网络开销、缓存命中率这些细节。每一步都有成本，建议按业务价值倒推目标，而不是按技术野心定目标。
3. 实操过程：从零搭一条实时智能链路
3.1 选定场景与指标
这里我拿一个最常见的电商场景举例：实时推荐与优惠券策略。用户在 App 里浏览商品、点击、加购，系统要根据最近几分钟的实时行为，结合库存和价格变化，在他进入下一个页面或发起下单动作之前，快速算出一个推荐排序和优惠方案。
这种场景的实时性价值特别明显：用户刚看了一双鞋，你立刻在下一个推荐位推同品牌或搭配商品，转化率通常比推热门但无关的商品高不少。但要做到这一点，行为事件从产生到特征生效，必须在 10 秒以内。上线前，我会先跟业务方把指标定清楚，不能只定一个“提升用户体验”这种没法验收的目标。
| 指标 | 目标值 | 计算方式 |
|---|---|---|
| 推荐点击率 | 相对提升 5% | A/B 实验对比 |
| 端到端推荐延迟 | p99 小于 200ms | 埋点上报时间差 |
| 行为事件到特征更新 | 小于 10s | 监控特征更新时间戳 |
指标定好之后，再决定实现范围。第一次做不要追求大而全，我建议只做三个特征：用户最近 5 分钟点击次数、当前浏览类目、商品实时库存状态。先让链路完整跑起来，再逐步加特征。
3.2 四步搭出最小链路
第一步：埋点事件接入。业务端把用户行为统一上报到 Kafka。事件格式要尽量规整，下面是一个最小示例。
{ "event_id": "uuid-xxxx", "event_time": "2025-05-18T10:24:03+08:00", "user_id": "u_10234", "item_id": "i_88901", "action": "click", "scene": "home_feed", "properties": { "category": "手机", "sku_price": 3999 } }topic 建议按业务域拆分，比如user_behavior、order_event、inventory_change，不要把所有事件塞进一个大 topic。分区键用user_id，保证同一个用户的实时行为按顺序进入同一个分区，窗口计算才能得到正确顺序。
第二步：Flink SQL 实时特征计算。我用 Flink SQL 做滑动窗口点击次数统计，样例大致长这样。
CREATE TABLE user_behavior ( event_time TIMESTAMP(3), user_id STRING, item_id STRING, action STRING, WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'user_feature_job', 'format' = 'json' ); CREATE TABLE user_feature_sink ( user_id STRING, window_start TIMESTAMP(3), click_cnt_5m BIGINT, PRIMARY KEY (user_id, window_start) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka', 'topic' = 'user_feature', 'properties.bootstrap.servers' = 'localhost:9092', 'key.format' = 'json', 'value.format' = 'json' ); INSERT INTO user_feature_sink SELECT user_id, TUMBLE_START(event_time, INTERVAL '5' MINUTE), COUNT(*) AS click_cnt_5m FROM user_behavior WHERE action = 'click' GROUP BY user_id, TUMBLE(event_time, INTERVAL '5' MINUTE);这里有两个细节特别重要。一是 watermark 的设置，它决定了你怎么处理迟到的数据，设置太短会丢数据，设置太长会加大延迟，后面第四章我会展开讲。二是流任务的状态管理，用户量和窗口数量会持续增长，必须为状态配置合理的 TTL，避免状态无限膨胀把任务拖垮。
第三步：特征写入在线存储并提供特征服务。流计算结果从 Kafka 落到 Redis，我通常用这样的 key 结构：usr_feat:{user_id}，类型用 Hash，字段就是特征名，比如click_cnt_5m、cat_1h。TTL 可以根据业务需要设置，营销场景一般在 2 到 4 小时，时间太长特征早就失去时效性，时间太短冷启动用户一多就会频繁回源离线库。
特征服务的读取逻辑要负责兜底。
def get_user_features(user_id: str) -> dict: key = f"usr_feat:{user_id}" data = redis.hgetall(key) if not data: # 实时特征没有，回退到离线画像，保证冷启动用户也有基础特征 data = offline_profile_service.get(user_id) return data冷启动是很多团队会忽略的问题。用户第一次访问时，实时特征是不存在的，如果你直接返回空，模型就只能瞎猜。正确的做法是给在线特征提供“默认值”或“离线兜底”，并且确保兜底策略跟训练时的缺失值填充策略一致，否则又会出现离在线不一致。
第四步：模型服务与规则兜底。模型服务拿到特征服务拼好的特征，做推理打分。
def rank(request): user_features = feature_service.get_user_features(request.user_id) item_features = feature_service.get_item_features(request.item_id) stock_status = inventory_service.get(request.item_id) if stock_status == 0: return None # 无库存直接不推荐 inputs = assemble_model_input(user_features, item_features) score = model.predict(inputs) strategy = select_strategy(user_features, request.scene) return {"score": score, "strategy": strategy}这里一定要有规则兜底。模型打分不是什么时候都可靠，比如刚上线时部分特征覆盖度低、模型对某些长尾商品完全没见过，这些情况要提前设计好降级策略，宁可返回一个保守的通用结果，也不能把空白内容推给用户。在线模型服务还要考虑模型版本管理，不能上线一个坏模型导致全站推荐效果突降而没法快速回滚。
3.3 上线前的验证与放量
链路搭好之后，不要直接拉全量流量，我按下面顺序做验证。
先做离线回放验证。拿最近一天的历史日志，喂给流计算任务回放，把计算出来的特征和离线批计算的特征做对比，看分布是否一致。这一步能发现窗口逻辑、时区、字段解析的差异。接着做影子部署，把线上真实流量复制一份打到新链路，但决策结果不真正生效，只记录模型分数和推荐结果，用来跟线上现有方案做离线对比。然后再灰度放量，从 5% 开始，逐步到 10%、30%、50%，每个阶段都盯住点击率、转化率和延迟指标。
放量期间还要关注一个容易被忽略的信号：用户反馈的异常率。如果推荐结果里频繁出现已下架商品、价格对不上、优惠条件不满足，说明特征链路里库存和价格字段的更新不及时。这类问题通常不会直接让模型分数崩掉，但会实实在在地伤害用户体验。
4. 常见问题与排查技巧实录
4.1 数据迟到和乱序：watermark 设置不合理
我第一次做实时特征计算，就踩过数据乱序的坑。移动端用户经常在弱网环境下操作，行为事件可能延迟几秒甚至几分钟才上报到服务端，如果流计算窗口按事件时间聚合，处理稍慢的事件就会被分到错误的窗口，或者直接被丢弃，导致实时特征偏小、偏旧。
排查这个问题，先在 Flink 的任务监控页面看两个指标：currentInputWatermark和事件时间延迟。如果 watermark 长期滞后于真实时间，说明设置的延迟容忍度不合理。我现在的做法是：对用户行为这类允许一定程度延迟的场景，watermark 设置成 30 到 60 秒，并配合 allowed lateness 让迟到事件有机会修正已经发出的窗口结果。
CREATE TABLE user_behavior ( event_time TIMESTAMP(3), user_id STRING, WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND )同时，写出到 Redis 的写入操作要设计成幂等，同一个用户同一个窗口的指标被迟到事件更新时，直接覆盖旧值，而不是累加。我曾经见过有人用 Redis 的INCRBY累加窗口计数，结果迟到事件重放后计数直接翻倍，这个坑排查了很久才发现。
4.2 训练与线上特征不一致
这是生产环境最隐蔽的问题，症状很典型：离线离线 AUC 很高，线上 A/B 实验却没有提升，甚至负向。我总结了几类高频原因，做成一个速查表：
| 检查项 | 可能原因 | 排查方法 |
|---|---|---|
| 特征定义不一致 | 离线在线各写一套逻辑 | 从 Feature Store 统一读取定义 |
| 特征时间穿越 | 用了决策时刻之后的数据 | 确认 point-in-time join |
| 缺失值策略不一致 | 线上返回空，训练用了均值填充 | 统一缺失值策略 |
| 字段类型不一致 | 离线是 int，线上接口返回 string | 对比样本和线上请求日志 |
| 分布漂移 | 线上特征值范围变了 | 监控特征分位数和 PSI |
排查这类问题，最直接的手段是抽样线上请求日志，把线上拼装出来的特征打印下来，再跟离线样本里同用户、同时段的特征做对比。两边的 min、max、均值、空值比例只要有一个明显对不上，就顺着链路往下查。不要靠感觉猜，一定要打日志看数据。
4.3 在线推理性能瓶颈
实时数据智能对延迟的要求很苛刻，推荐场景 p99 要到 200 毫秒以内，风控场景更严格。性能问题出现时，不要一上来就怪模型太慢，我见过大部分延迟都浪费在特征读取和序列化上。
有一次我优化一个推荐服务，发现 p99 一直压不下去，逐段打点之后发现特征服务里两个 Redis 请求是串行发的，用户特征读一次、商品特征读一次，加起来快 30 毫秒。改成 Redis pipeline 一次批量读取后，直接降到 12 毫秒。还有一次是模型打分用 JSON 传参，每次请求序列化开销很大，后来换成 protobuf 格式，整体延迟又降了一截。下面是典型的耗时拆解参考：
| 环节 | 参考耗时 | 优化手段 |
|---|---|---|
| 特征读取 | 15~30ms | Redis pipeline、本地缓存 |
| 特征拼接与校验 | 3~8ms | 预编译特征顺序 |
| 模型推理 | 10~50ms | 批处理、模型裁剪或量化 |
| 结果下发 | 2~10ms | 精简响应字段、压缩 |
缓存是一种有效手段，但要注意新鲜度。特征本地缓存设置几十毫秒到几百毫秒的 TTL 完全可行，既不影响模型看到的数据新鲜度，又能显著降低 Redis 压力。如果业务对新鲜度极其敏感，比如价格和库存，我建议这类特征不做本地缓存，直接从在线存储读取，把缓存只留给用户行为这类允许小延迟的特征。
4.4 成本曲线失控
流计算是 7×24 小时跑的，成本跟离线批任务不是一个量级。我见过一个团队上线实时推荐链路后，第一个月成本比预估高了三倍，原因很朴素：窗口计算开得太多，加上状态没有设置 TTL，任务内存一直在涨。
控制成本有几个务实的手段。第一，不要所有特征都实时算，先分辨哪些特征真正需要秒级更新，哪些其实小时级就够了；对后者，用定期批刷新，成本会低一个数量级。第二，给流任务的状态设置合理 TTL，并且定期做 checkpoint 清理，避免状态膨胀。第三，Redis 集群按业务优先级规划容量，冷门特征设置较短的 TTL，缓解内存压力。第四，模型服务优先考虑 CPU 推理，只有业务量确实大或者模型必须用 GPU 时才上 GPU 服务，因为 GPU 实例的空闲成本很惊人。
还要提醒一点：成本问题要前置到技术评审阶段，而不是上线之后才看账单。上线前就估算出每个用户每日请求数、事件量、状态大小、推理 QPS，乘以资源单价，让业务方清楚知道这个实时智能功能是有持续运营成本的。否则功能上线后因为成本被砍掉，前面所有功夫都白费。
5. 选型与工程路线的几条经验
5.1 组件选型：少即是多
聊到实时数据智能的架构，大家很容易被新技术吸引，最后造出一个特别复杂的系统。我的原则始终是：用团队最熟悉的技术，先把最小闭环跑通，再考虑替换升级。
如果团队熟悉 Flink，就统一用 Flink 做流式计算；如果团队规模小、事件量不大，Spark Structured Streaming 也完全够用，不一定非要上 Flink。在线存储首选 Redis，键简单、生态成熟、排查问题容易，只有当特征维度特别复杂、需要大规模倒排查询时才考虑 HBase。消息总线用 Kafka 就足够了，Pulsar 的优势在多租户和某些高级特性上，对大多数 AI 应用场景并不是必需品。服务化方面，先用简单的 Python 服务加模型文件部署，当请求量上来之后再引入专门的模型服务框架。整体架构每多一个组件，就多一层排查链路和运维成本，选型的最高优先级永远是“当前阶段够用且团队能维护”。
5.2 让算法、数据和平台团队在同一张蓝图上协作
实时数据智能链路的责任边界特别容易模糊：算法团队负责模型，数据团队负责加工特征，平台团队负责资源，业务方最后看到的却是整体效果。一旦线上出了问题，三个团队互相推，谁都能找到“不是自己这边的问题”的理由。
我建议从一开始就设立一个端到端 Owner，不管是数据团队还是算法团队的人，他必须对整体延迟和业务指标负责，有权调动三个团队的资源。同时，特征定义和模型版本要一起纳入发布流程，每次模型发版时，同时核对它依赖的特征是否已经上线、特征取值逻辑是否一致。我在团队里还规定了故障演练制度，每个月主动把 Kafka 集群搞挂一次，或者把某个特征服务停掉，观察模型服务的降级逻辑是否生效。这些演练看着折腾，但线上真出故障的时候，能帮你省下好几个小时的排查时间。
5.3 演进路径：从批量到实时的三步走
最后聊一下演进路线。我强烈不建议一个团队第一次做实时数据智能就全线铺开，更稳的路子是分三步走。
第一步，先把 T+1 的离线特征补齐，把离线链路和评估体系做扎实。这一步看起来不性感，但它是后面所有工作的基准线。第二步，做“半实时”混合链路：用户画像、静态商品属性继续用离线特征，只有行为序列、库存、价格这几个高时效特征走实时链路。这种方式改造成本小，但已经能覆盖大部分业务价值场景。第三步，当实时链路的稳定性经过验证后，再逐步扩大实时特征范围，再去考虑在线学习和模型自动回流这类更高级的能力。
这个顺序背后有一个简单的逻辑：每一步都有明确的业务增量和验收指标，风险可控。如果反过来，一上来就追求全链路毫秒级实时，大概率会陷入成本失控、问题排查困难的泥潭，最后连最基础的业务价值都交付不了。
我个人在实际操作中的体会是，实时数据智能不是一个“做完就结束”的项目，而是一个持续调优的工程体系。你每把延迟压缩一个数量级，业务方就会提出新的数据需求，而每一次新增数据需求，都可能引发链路稳定性的新问题。所以不要怕慢，先把端到端时延从一天压到一小时，拿到业务增量，再压到分钟、秒级，每一步都走扎实。
最后再分享一个小技巧：上线之后一定要做事件漂移监控，把关键特征的分布变化、模型分数的 PSI 变化都接到告警群里。很多线上事故最早的信号不在系统报错里，而在数据里。某个特征的分布突然偏移，往往意味着上游埋点被改、业务规则变化或者数据源出了故障，这些都是靠模型算法本身发现不了的。把数据监控守好，你的实时数据智能链路才算真正有了自我保护能力。
