1. 先把“算子”这件事讲透：数据管道的真实结构

先给结论：绝大多数数据管道翻车，不是翻在“跑不通”，而是翻在算子这一层设计得含糊。所谓算子，就是管道里一次最小的数据加工动作，读取一条记录、解析一个字段、过滤一批脏数据、按时间窗口算个指标，每个动作都是一个算子。把这些算子按依赖顺序串起来，才是我们嘴里的数据管道。刚接手一个需求的时候，我一般不看需求文档写得多漂亮，先问一句“你这批数据从哪来、中间要过几手、最后落到哪”，这句话基本就把整条管道的算子骨架描出来了。

这篇文章我打算聊的是：一条数据管道应该怎么从零构建，算子要按什么粒度切、每个算子内部要注意什么、上线之后怎么盯。适合刚入行的数据开发照着抄骨架，也适合已经能写管道但总被回压、数据倾斜、重复数据折腾的中级同学对照排查。文中所有场景我都做了脱敏处理，代码用通用写法，换成你手头的引擎基本能平移。

我见过最典型的一种反例：一个同学把“清洗、去重、关联维表、聚合、落库”全塞进一个超大算子，看起来代码只有两百行很清爽，结果一跑就卡，一卡就找不到哪一步慢。原因很简单，你没法知道时间花在解析 JSON 上还是花在查维表上。算子的第一条纪律就是可观测，一个算子只干一件事，出了事才知道该修哪一段。

1.1 一条管道到底是哪几段：源头、算子、汇聚

不管业务多花哨，一条管道拆到骨头里就三段。源端负责把数据拿进来，可能是业务库的变更日志、埋点上报的日志流、第三方推过来的文件、或者一批历史数据；算子段负责所有加工，清洗、补全、聚合、关联、脱敏、分流；汇聚端负责把结果写出去，写宽表、写指标库、写一个供下游订阅的主题、或者写文件给离线跑批。

这个三段划分看着废话，但它决定了你排查问题的方向。源端出问题，症状通常是“延迟持续上涨但吞吐没降”，因为数据干脆没进来；算子段出问题，症状是“某几个并行实例的指标明显偏高”，通常是倾斜或某个外部依赖变慢；汇聚端出问题，症状是“处理很快但端到端延迟高”，因为数据堵在写出这一关，尤其是批量写和事务提交的时候。

我习惯在动手写代码之前先画一张三段图，把每段的输入输出字段列到字段级别。这一步花二十分钟，能省后面两天。原因很实在：算子之间的契约如果不先定死，你写到第三个算子的时候一定会发现上游少给了一个字段，然后回头改，改完上游的清洗逻辑又要重测，连锁反应。

1.2 为什么“算子”是决定成败的那一层

源端和汇聚端的技术选型基本被环境绑死了，你能折腾的空间就在算子层。而算子层的复杂度来自三个地方：状态、顺序、并行。

无状态算子最好写，来一条处理一条，map、filter、简单字段变换都是这一类，扩并行度直接加实例就行。有状态算子就麻烦了，去重要知道“这条之前来过没有”，窗口聚合要知道“这一分钟的数据收齐了没有”，关联维表要知道“这张维表的最新版本是什么”。状态意味着你不能再随心所欲地重跑，得考虑检查点、恢复位点、状态的过期清理。

顺序这件事更隐蔽。很多人默认“消息是按时间顺序来的”，实际上多分区并行读取之后，到达顺序基本是乱的，尤其是跨分区。所以凡是涉及“先来后到”的业务逻辑，比如“以最后一次修改为准”，你必须显式地用事件时间加版本号去裁决，不能靠到达顺序。我踩过这个坑：两个更新先后写入不同分区，下游按到达顺序覆盖，结果最终状态是旧值，排查了大半天。

并行则带来分区和倾斜的问题。同一个 key 必须落到同一个并行实例上，聚合结果才正确，这就是分区键的意义。分区键选错，要么结果错，要么某些实例被压垮。

1.3 批、微批、流三种架构的取舍

选架构这件事，我一般用一张表跟需求方对齐，比讲半小时原理管用。核心是三列：数据延迟要求、结果准确性要求、运维成本承受度。

我一般的判断标准是：业务方能不能接受“最坏情况下晚 15 分钟看到数据”。能接受，就别上持续流处理，定时批处理加幂等重跑，运维成本差一个量级。不能接受，再看是不是真的需要亚秒级，很多“实时”需求其实是“一分钟内”，微批就够了。

这里有个经验：别用架构的复杂度去换一个业务根本感知不到的延迟。我见过一个团队为了“实时”两个字上了全套流处理，结果下游消费方每五分钟才拉一次数据，前面所有的低延迟设计全部浪费，反而多背了一套状态存储和检查点机制的运维包袱。

2. 设计阶段：把需求翻译成算子的四步拆解

设计阶段最容易犯的错是拿到需求就开始写代码。我一般的顺序是：先定数据契约，再定幂等策略，再算资源，最后才决定要不要上状态。这四步的顺序不能反，反了就要返工。

2.1 数据契约先行：字段、类型、语义三件套

数据契约不是写个字段名列表就完事，要写清三样东西：类型、可空性、语义。类型好理解，字符串还是长整型；可空性指的是这个字段允不允许缺失，缺失时用什么默认值填充；语义最容易漏，比如amount到底是分还是元、ts是事件发生时间还是入库时间、status的枚举值分别代表什么。

我习惯把契约写成一份结构化的描述文件放在代码仓库里，跟算子代码一起做版本管理。这么做的好处是，上游改字段的时候你能在评审阶段就发现，而不是等到跑起来发现一堆空指针。下面是我常用的一个极简契约写法，用 YAML 描述，字段语义写在注释里：

# contract/order_events.yaml name: order_events version: 3 fields: event_id: {type: string, nullable: false} # 全局唯一，用于去重 user_id: {type: string, nullable: false} # 用户标识，脱敏后存储 event_time: {type: long, nullable: false} # 事件发生时间，毫秒时间戳，UTC ingest_time: {type: long, nullable: false} # 入库时间，毫秒时间戳，UTC amount_cent: {type: long, nullable: true} # 金额，单位：分；退款为负 status: {type: string, nullable: false} # paid/refunded/cancelled ext: {type: string, nullable: true} # 原始扩展 JSON，保留原样

这份文件的价值在评审会上体现得最明显。有次我把amount_cent的单位写进契约，需求方当场发现他们以为单位是元，差了一百倍。这种错误如果跑到线上才发现，修复成本是把已经写进指标库的历史数据全部重算。

注意：event_time和ingest_time一定要分开。只留一个时间字段的管道，后面做迟到数据处理、做重跑补数的时候一定会痛。

2.2 幂等与去重：算子设计里最容易被忽略的一环

数据管道一定会重跑，这不是“会不会”的问题，是“什么时候”的问题。消息重投、任务失败重启、人工补数、上游修正历史数据，每一次都要求你的算子能承受重复输入。所以从设计第一天起，每个写出算子都必须是幂等的。

幂等的实现路径我一般分三档，按成本从低到高：

第一档是主键覆盖写。目标存储有主键约束，写入用覆盖语义而不是追加语义，同一批数据重跑多少遍结果都一样。这档最省心，前提是目标端支持主键。

第二档是去重键 + 状态。在算子内部维护一个去重集合，key 用业务主键或者event_id，命中就丢弃。这档要管状态大小，得给 key 设过期时间，否则状态会无限膨胀。

第三档是版本号裁决。同一主键可能有多次更新，用版本号或者更新时间戳比较，只保留最新的。这档解决的是“覆盖顺序”问题，比单纯去重更严格。

实际项目里通常是第二档加第三档组合。我给一个去重算子的通用骨架，用 Python 伪代码写，逻辑跟具体引擎无关：

# dedup_operator.py class DedupOperator: def __init__(self, state, ttl_seconds=7 * 86400): self.state = state # 外部状态存储，支持 get/put/expire self.ttl = ttl_seconds def process(self, record): key = record["event_id"] # 第一层：按 event_id 去重，防止消息重投 if self.state.get("seen:" + key): return None # 第二层：按业务主键比版本，防止乱序覆盖 biz_key = "ver:" + record["user_id"] + ":" + record["order_id"] last_ver = self.state.get(biz_key) if last_ver is not None and last_ver >= record["event_time"]: self.state.put("seen:" + key, 1, ttl=self.ttl) return None self.state.put(biz_key, record["event_time"], ttl=self.ttl) self.state.put("seen:" + key, 1, ttl=self.ttl) return record

这段代码有两个细节值得说。一是seen:和ver:用了不同的前缀分开存，方便后面按前缀做统计和清理；二是先去重再比版本，两个状态都要写 TTL，不然状态只增不减。

提示：去重状态的 TTL 必须大于“最大可能迟到时间”。如果你的数据源允许迟到七天，TTL 设成一天，那么第七天到的重复数据就会被当成新数据，直接穿透。

2.3 并行度、分区键与资源估算

并行度这个参数，新手最容易犯两种错：一是拍脑袋写个 10，二是照抄别的任务的配置。并行度的正确算法是从吞吐倒推。

我给一个具体的估算例子。假设某业务日增事件 2 亿条，平均单条 1.2KB，峰值流量是均值的 3 倍。先算平均 QPS：200,000,000 / 86400 ≈ 2315 条/秒。峰值 QPS ≈ 6944 条/秒。单核处理这类带 JSON 解析的算子，实测大概 6000 到 8000 条/秒，保守取 6000，那么处理需要 ≈ 1.16 核。加上反序列化、序列化、网络、检查点的开销，实际留 3 到 4 倍余量，并行度取 4就够，取 16 属于纯浪费，还会因为小批量写出产生一堆小文件。

带宽也要算一遍：6944 × 1.2KB ≈ 8.3MB/s，这个量级单机网络完全扛得住。落库存储估算：2 亿 × 1.2KB = 240GB/天，列式压缩按 3 倍算，实际落盘约 80GB/天，按月算 2.4TB，这个数字要提前跟存储方对齐，别上线一周把磁盘写满。

分区键的选择原则只有一条：所有需要聚合的维度，必须在分区键里。按用户维度聚合，分区键就用用户标识；按订单聚合，就用订单标识。如果既要按用户聚合又要按商品聚合，那就得拆成两条管道，各写各的结果，不要试图在一个算子里用一套分区键满足两种聚合。

这里有个坑我必须提醒：别用高基数且分布不均的字段当分区键。用“城市”当分区键，头部城市的数据量可能是尾部的一百倍，那个并行实例会被打死，其他实例闲得发慌。遇到这种情况，要么加盐打散做两阶段聚合，要么把分区键换成更均匀的字段。

2.4 状态与窗口：什么时候必须上状态算子

状态不是越少越好，也不是越多越好，关键看业务语义需不需要“记住过去”。我把常见需求分成三类，对应三种处理方式：

完全无状态：单条记录能独立算完的，比如字段脱敏、单位换算、格式校验、打标签。这类直接 map/filter，不要引入任何状态。

窗口内状态：需要“一段时间内”的聚合，比如“每分钟的下单量”“最近五分钟的失败率”。这类用窗口算子，状态大小跟窗口长度和数据速率成正比，可控。

跨窗口长期状态：需要“历史至今”的累计，比如“用户生命周期总消费”。这类要小心，状态会无限增长，通用做法是用外部存储做累加，算子内部只做增量计算和批量提交，而不是把全量状态放在算子内存里。

窗口这里必须讲清楚事件时间和处理时间的区别。事件时间是数据里带的发生时间，处理时间是算子拿到这条数据时的机器时间。用事件时间做窗口，才能得到可重放、可重算的正确结果；用处理时间做窗口，重跑一次结果就变一次，对不上账。我习惯一律用事件时间做窗口，配合水位线控制“什么时候认为这一批收齐了”。

水位线可以理解成一句声明：“我认为所有时间戳早于 X 的数据都已经到了”。设得太松，结果出得慢；设得太紧，迟到数据全被丢掉。常见做法是允许 30 秒到 5 分钟的迟到，具体看你的数据源波动情况，同时把超期的迟到数据旁路输出到一个单独的位置，方便事后核对丢了多少。

3. 动手实现：一条可跑通的管道与关键算子代码

这一节我把一条完整的管道从骨架到细节写一遍，场景是“订单事件流加工成用户维度的分钟级指标”，代码用通用写法，引擎相关的部分我会标注出来。

3.1 骨架搭建：源、变换、汇聚的代码

我把管道骨架分成四层：source 层只负责读和反序列化，transform 层是各种算子按顺序串起来，sink 层只负责写，config 层放所有可调参数。分层的好处是换引擎的时候只改 source 和 sink，算子逻辑不用动。

# pipeline.py from operators import ParseOp, ValidateOp, DedupOp, EnrichOp, WindowAggOp, SinkOp def build_pipeline(cfg, state): ops = [ ParseOp(cfg["schema"]), # 字节流 -> 结构化记录 ValidateOp(cfg["rules"]), # 契约校验，不合规的旁路输出 DedupOp(state, ttl_seconds=cfg["dedup_ttl"]), EnrichOp(cfg["dim_table"], cfg["dim_refresh_sec"]), WindowAggOp(cfg["window_sec"], cfg["allowed_lateness_sec"]), SinkOp(cfg["sink"], batch_size=cfg["sink_batch"], flush_ms=cfg["flush_ms"]), ] return Chain(ops) if __name__ == "__main__": cfg = load_config("config/pipeline.yaml") state = StateBackend("checkpoint/pipeline") pipe = build_pipeline(cfg, state) pipe.run(parallelism=cfg["parallelism"], checkpoint_ms=cfg["ckpt_ms"])

这里Chain的作用是把算子顺序串起来，中间任何一步返回None就表示丢弃，不再往下传。这样“过滤”这个语义就统一了，不用每个算子各写一套判断。

配置我单独抽出来，方便按环境覆盖：

# config/pipeline.yaml schema: contract/order_events.yaml parallelism: 4 ckpt_ms: 60000 dedup_ttl: 604800 # 7 天 dim_refresh_sec: 300 # 维表每 5 分钟刷新一次 window_sec: 60 allowed_lateness_sec: 120 # 允许迟到 2 分钟 sink_batch: 500 flush_ms: 200 sink: type: overwrite_upsert primary_key: [user_id, window_start]

checkpoint_ms设 60 秒是个折中。设太短，检查点开销占比高，吞吐下降；设太长，故障恢复时要重放的数据多，恢复慢。60 秒在大多数场景下是舒服的值，如果状态很大（超过几十 GB），改成增量检查点配 120 秒。

3.2 清洗与校验算子：那些看起来不起眼但天天出事的地方

清洗算子看着简单，实际上八成的数据问题都出在这一层。我按踩坑频率排了个序：时间戳时区、空值语义、字符串编码、数值精度、嵌套结构解析。

时间戳是最容易出事的一个。上游给的时间戳如果没带时区，你按本地时区解析，跨时区场景下一定错，而且错得很隐蔽，因为只差几个小时，指标看起来“差不多但不对”。我的做法是强制契约里声明 UTC，解析后统一转成毫秒整数，所有算子内部只用整数时间戳，只在最终展示时做时区转换。

空值这块要区分“字段缺失”和“字段值为空字符串”，业务语义可能完全不同。比如ext字段缺失表示上游没传扩展信息，值是空字符串可能表示传了但是内容为空。校验算子要把这两种情况分开处理，别一概而论都填默认值。

编码问题值得单独说。清洗时遇到非法字节序列，直接抛异常会把整条管道打挂，更稳的做法是做替换并计数，同时把原始字节旁路存下来。这样既保证了管道不中断，又能事后追查是哪批数据有问题。

# operators/validate.py class ValidateOp: def __init__(self, rules): self.rules = rules self.rejected = Counter() def process(self, rec): for field, rule in self.rules.items(): val = rec.get(field) if val is None: if not rule["nullable"]: self.rejected["missing:" + field] += 1 return None # 必填缺失，整条丢弃并计数 rec[field] = rule.get("default") continue if rule["type"] == "long" and not isinstance(val, int): try: rec[field] = int(val) # 宽容处理：能转就转 except (TypeError, ValueError): self.rejected["badtype:" + field] += 1 return None if "enum" in rule and rec[field] not in rule["enum"]: self.rejected["badvalue:" + field] += 1 return None return rec

这段代码有个刻意的设计：数值类型的校验做宽容转换而不是直接拒绝。上线初期上游的类型经常不稳定，字符串和数字混着来，宽容转换能少掉很多无效告警。但枚举值校验必须严格，因为枚举错了说明上游语义变了，这种要立刻发现。

注意：被丢弃的数据一定要打点计数并且旁路落盘，不然你会遇到“数据量怎么少了一截但没人知道少在哪”的经典问题。

3.3 聚合与窗口算子：水位线、迟到数据、触发时机

窗口聚合是整条管道里最容易写错的部分。我把关键点拆成四个：窗口类型、触发时机、迟到处理、结果更新。

窗口类型分滚动窗口和滑动窗口。滚动窗口不重叠，适合做“每分钟一个点”；滑动窗口重叠，适合做“最近五分钟、每三十秒滚动一次”的移动平均。滑动窗口的状态开销是滚动窗口的好几倍，用之前先确认业务真的需要滑动。

触发时机由水位线决定。水位线推进到“窗口结束时间 + 允许迟到时间”之后，窗口触发计算并输出。这里有个细节：很多引擎支持“提前触发”和“延迟更新”，也就是窗口到点先输出一个近似结果，等迟到数据到了再更新一次。如果你的下游是覆盖写，延迟更新完全没问题；如果是追加写，就得想清楚同一条结果怎么处理两次写。

迟到数据的处理我给两条路：一是允许迟到，把窗口保留时间设得足够长，代价是状态变大、结果延迟；二是旁路输出，超期的数据写到单独的通道，事后手工核对。我一般两条都用，主链路设 2 分钟容忍，超期的旁路存着。

# operators/window_agg.py class WindowAggOp: def __init__(self, window_sec, lateness_sec): self.window_sec = window_sec * 1000 self.lateness = lateness_sec * 1000 self.windows = {} # {window_start: {user_id: acc}} def process(self, rec): ts = rec["event_time"] wstart = ts - (ts % self.window_sec) if ts < self._watermark() - self.lateness: return ("late", rec) # 超期，走旁路 acc = self.windows.setdefault(wstart, {}) slot = acc.setdefault(rec["user_id"], {"cnt": 0, "amt": 0}) slot["cnt"] += 1 slot["amt"] += rec.get("amount_cent", 0) return None

实际落地时，窗口的清理要跟水位线联动，水位线推进过的窗口及时从状态里删掉，否则跑一周状态能涨到几十 GB。我踩过一次：窗口状态忘了删，检查点越来越大，从几百兆涨到 20GB，恢复一次要七八分钟，最后只能停机手工清状态。

3.4 给算子加埋点：没有指标的管道等于没有管道

埋点这部分很多人偷懒，觉得“跑起来没报错就行了”。等你半夜被叫起来处理延迟告警，没有指标就只能靠猜。

我一般在每个算子的入口和出口各打三个数：输入条数、输出条数、处理耗时分布。输入输出条数一对比就知道这个算子丢掉了多少数据，耗时分布能定位到具体是哪一步变慢。另外再加两个全局指标：端到端延迟（当前处理时间减去数据事件时间）和状态大小。

这张表里我最看重的是“输出/输入比”。这个比值突然变化，往往比延迟告警更早发现问题。有次上游改了一个字段的默认值，导致过滤条件命中率从 98% 掉到 60%，延迟指标一切正常，就是这条比值先报的警。

3.5 调度与依赖编排：重跑幂等是底线

管道跑起来之后，运维的重头戏就是调度。批处理管道要靠调度器串依赖，流处理虽然常驻，但补数、重跑、版本升级都离不开调度。

依赖编排的核心原则是：上游没成功，下游绝不启动，同时下游重跑必须能覆盖上游修正的数据。第二点经常被忽略。上游发现了脏数据，修完重跑，下游如果不跟着重跑，两边就对不上账。

我一般的做法是给每条管道标记好“依赖的数据分区”，重跑时按分区粒度触发下游。下面是一个通用的重跑脚本，核心是先把目标分区标记成“处理中”，再跑，成功才标记“完成”，失败就保持“处理中”，避免下游误读半成品数据：

#!/usr/bin/env bash set -euo pipefail PIPE="order_metrics" DT="${1:?usage: rerun.sh <yyyy-mm-dd>}" mark_status() { ./ctl status --pipe "$PIPE" --dt "$DT" --state "$1"; } mark_status running if ./ctl run --pipe "$PIPE" --dt "$DT" --overwrite; then mark_status done ./ctl trigger-downstream --pipe "$PIPE" --dt "$DT" else mark_status failed exit 1 fi

--overwrite这个参数必须有，它保证重跑是覆盖而不是追加。这个参数如果实现不了，说明你的 sink 层设计有问题，得回去改。

4. 上线之后：监控、回压与故障处理实录

上线只是开始。数据管道的生命周期里，稳定运行的时间远小于修问题的时间，所以这一节的篇幅我给得最足。

4.1 监控该看哪几个数，怎么看出问题

监控面板不要堆几十个图，看不过来。我一般只留四个图，从上到下回答四个问题：数据进得来吗、处理跟得上吗、结果对不对、资源够不够。

数据进得来吗，看源端速率和处理速率两条线。两条线应该贴合，如果处理速率长期低于源端速率，说明在积压。

处理跟得上吗，看端到端延迟的 P50 和 P99。P50 平稳但 P99 抬头，通常是少数慢请求或者倾斜；两个都抬头，通常是整体资源不够或者下游变慢。

结果对不对，看输出条数、关键指标的空值率、枚举值的分布。这三个是数据质量的哨兵，比业务方的投诉早得多。

资源够不够，看 CPU、内存、状态大小、检查点耗时。检查点耗时突然从 2 秒涨到 20 秒，基本就是状态膨胀或者外部存储变慢。

4.2 典型故障排查速查表

下面这张表是我这几年攒下来的，按症状查因，命中率还挺高：

我挑两条重点说。“少数实例 CPU 打满”这条最容易被误判成资源不够，然后加机器，加完还是打满。先看分区键分布，如果某个 key 占了 40% 的数据量，加多少机器都没用，因为同一个 key 只能在一个实例上处理。

“恢复后数据错乱”这条最凶险，因为它不告警，等业务方发现账对不上才暴露。根因通常是状态没纳入检查点，恢复之后去重集合空了，历史数据全部重新算一遍。

4.3 回压：管道在“喊停”的时候你要听懂

回压是流处理里的常见现象，说白了就是下游处理不过来，上游被反推着降速。表现出来的症状是源端消费速率自动下降，看起来像“上游没数据”，其实是下游堵了。

排查顺序我是固定的：先看下游 sink 的写入耗时，再看算子链里最慢的那一环，最后看是否有同步阻塞调用。常见的原因有四个：目标库单条写入太多没批量、维表查询是同步远程调用、某个算子做了全量排序、检查点期间和正常处理抢 IO。

我遇到过最典型的一次：sink 层为了“保证实时性”设了flush_ms=10且batch_size=1，等于每条都单独提交一次事务。目标库每秒只能接住两千次提交，而管道每秒产出七千条，直接回压。改成batch_size=500、flush_ms=200之后，端到端延迟反而从 8 秒降到 1.5 秒，因为事务开销摊薄了。

提示：批量写和实时性不是对立的。批量参数调到“延迟增量小于业务容忍度”即可，不要为了心理上的实时把批量关掉。

4.4 变更与回滚：怎么改管道而不出事

管道要改是必然的，关键是改的时候别把线上打挂。我的流程是四步：新增字段走加法、算子逻辑走灰度、状态结构变更走双跑、回滚方案先写好。

新增字段走加法，意思是只加不删不改类型。删字段和改类型都要走“新老并存一段时间”的过渡期，等所有下游都切完再删。

算子逻辑灰度，指的是同一份输入同时跑新旧两套逻辑，结果写到不同位置，对比一段时间没问题再切。这个成本看着高，但对于核心指标管道是值得的。

状态结构变更最麻烦，因为检查点里的状态格式变了，老检查点恢复不了。稳妥做法是新起一套任务双跑，追上进度后切流量，老任务保留一段时间以便回滚。

回滚方案必须在变更之前写好，写清楚“怎么退回上一版本、退回后状态怎么处理、已经写出去的数据要不要清理”。我见过太多“改完发现问题但不知道怎么退”的情况，最后只能停机几个小时手工修数据。

5. 实操心得与避坑清单

写到这儿，把一些零散但值钱的经验归拢一下，都是文档里不会写、只有踩过才知道的。

5.1 我踩过的几个坑

第一个坑：用处理时间做窗口。早期做实时指标，图省事直接用处理时间开窗。跑了一周业务方来问“为什么昨天下午三点的数据和今天重算的不一样”。因为重跑的时候处理时间变了，窗口边界跟着变，结果自然对不上。改成事件时间之后，重跑结果完全一致。这个教训让我从此在任何涉及聚合的场景都强制用事件时间。

第二个坑：去重状态没设 TTL。一开始想着“去重集合当然是越多越好”，结果状态一个月涨到 40GB，检查点每次要写 40GB，检查点间隔 60 秒根本写不完，任务一直在超时和恢复之间循环。加上 7 天 TTL 之后，状态稳定在 4GB 左右。

第三个坑：维表关联用同步远程调用。每条数据都去查一次维表，维表服务扛不住，延迟从 50ms 涨到 3 秒。改成定时全量拉取加本地缓存，五分钟刷新一次，延迟降到 5ms。维表变更不频繁的场景，本地缓存永远优于远程查询。

第四个坑：以为分区数可以随便改。流处理里分区数改了，分区键到实例的映射就变了，如果状态是按分区存的，改完恢复会错乱。要改分区数，得先停任务、做一次状态迁移、再启动，不能在线改。

5.2 命名与文档：半年后的你会感谢现在的你

管道的命名我坚持三个约定。算子名用“动词 + 对象”，比如ParseOrderEvent、DedupByEventId、AggByUserMinute，一眼看出干什么。配置项名带上单位，window_sec而不是window，flush_ms而不是flush，省掉无数“这个参数是秒还是毫秒”的追问。旁路输出统一加后缀，_rejected、_late、_dlq，扫一眼就知道哪些是异常数据通道。

文档我至少写三样：数据契约、算子依赖图、重跑手册。第三样最容易被忽略，但最救命。重跑手册要写清楚：哪些管道可以并行重跑、哪些有依赖顺序、重跑会不会影响下游、重跑期间要不要暂停下游消费。半夜三点被叫起来的时候，你不会想靠回忆来推理这些。

5.3 从小管道长成大管道的演进路线

最后聊个务实的：别一上来就设计一套能支撑全公司的数据平台。我见过太多团队在只有三条管道的时候就开始搭“统一数据平台”，搭了三个月，管道一条没上线。

务实的路线分三段。第一阶段，三条以内的管道，直接写脚本加调度器，配置写文件，够用。第二阶段，管道到十几条，抽出公共算子库和数据契约，统一埋点和告警。第三阶段，管道几十条、跨团队协作，才需要统一的元数据管理、血缘追踪、平台化调度。

每一段的判断标准很简单：当你开始重复写第三遍同样的代码时，就该抽公共库了；当你开始说不清某个指标是从哪条管道算出来的时候，就该上血缘了。别提前，也别拖。

我个人在实际操作中的体会是，数据管道这件事，八成的功夫在设计阶段，两成在写代码。设计阶段想清楚契约、幂等、分区、状态这四件事，后面就是按部就班地填代码和盯指标。反过来，设计阶段糊弄过去，后面就是无休止的排查、补数、改结构，而且每次改动都心惊胆战。慢一点设计，快一点上线，这个账怎么算都划算。