现在的数据团队几乎每天都会遇到一个矛盾业务方上午十点问你要昨天的数据你翻半天发现凌晨的批量任务卡在某个上游接口上数据到现在还没落地另一边实时大屏的指标又和离线报表对不上两边口径谁都说不清。这套问题的核心往往不是数据仓库建模不行也不是分析师的SQL写得差而是最前端的数据摄取Ingestion环节没有做扎实。Ingestion服务简单说就是把外部系统的数据导入到内部数据系统的这一段链路它决定了数据能不能进得来、以什么姿态进来、以及进来了之后下游能不能接得住。这篇内容我基于做Ingestion服务的实际经验重点聊聊Batch Ingestion批处理摄取和Streaming Ingestion实时摄取这两种模式的内部逻辑、选型思路、架构设计以及落地过程中的坑。适合正在搭建数据平台的工程师、刚接手数据接入任务的同学以及想搞明白“数据到底是怎么进数仓”的业务方。我会尽量把原理讲透把方案讲具体给出可以直接抄作业的设计思路。1. 先想明白Ingestion服务到底在解决什么问题很多人会把Ingestion和ETL混在一起这其实是两类东西。ETL里也有Extract抽取但那个抽取通常发生在数仓内部面向的数据源是业务库、日志、文件这些“已知对象”而Ingestion服务站在更靠前的位置它面对的是“外部世界”可能是合作伙伴的FTP、第三方API、线下系统的Excel导出文件甚至是一天到晚变动schema的数据库。也就是说Ingestion服务是整个数据体系的入口也是各种脏数据、怪数据、迟到数据集中爆发的地方。我通常把Ingestion服务要解决的问题拆成四块统一接入层屏蔽差异不同数据源的协议千奇百怪有JDBC、Kafka、HTTP、SFTP、S3、ODBC甚至还有上古的ODBC over MQ。Ingestion服务要把这些差异封装起来对外暴露统一的数据落地结果不让下游感知到上游的混乱。格式和类型的归一化上游给的时间可能是个字符串也可能是Unix时间戳金额可能带千分位逗号可能是个浮点字符串。Ingestion的目标是让数据流出这个服务时已经是一套清晰、自洽、符合内部规范的结构。数据的完整性和准确性保障数据在传输过程中会丢、会重、会乱序Ingestion服务要通过幂等设计、校验机制、重试策略来保证下游拿到的数据是“够用且正确”的。时效性的分级满足不是所有数据都要实时。订单表要分钟级财务对账文件可以T1埋点日志需要秒级。Ingestion服务要做的是提供不同时效等级的能力而不是一刀切。想明白了这四点设计时就不会一上来就扯Flink还是Spark而是先看你接入的数据源到底长什么样。2. 选批处理摄取还是实时摄取关键是看这几个变量Batch Ingestion和Streaming Ingestion不是简单的“快与慢”的关系它们从设计哲学上就是两套东西。我见过的失败项目里相当一部分是选了错误模式硬上比如把所有数据都改成实时结果成本和运维压力翻了好几倍或者对时效要求极高的场景还跑每小时批处理被业务方追着骂。下面这张表是我在做技术选型时常用到的对比维度对比维度Batch Ingestion批处理摄取Streaming Ingestion实时摄取典型处理单位一批文件 / 一段时间切片小时、天一条条事件 / 微批量窗口端到端延迟分钟级到小时级秒级到分钟级数据源形态文件、冷存储、数据库导出、夜间接口消息队列、日志流、CDC变更流、埋点事件成本特征波峰波谷明显空闲期资源可释放常驻计算资源成本稳定数据正确性保障重跑、全量校验比较方便需要靠状态、幂等、精确一次语义兜底运维复杂度相对较低重试和补偿好做高要关心水位、背压、故障恢复典型场景财务对账、历史数据回溯、大数据量入库风控、大屏监控、推荐特征、告警决策时我最看重的三个变量是数据源的能力、下游的消费方式、以及业务对延迟的真实容忍度。数据源本身不支持流式推送比如上游给的是每天凌晨生成的文件这时候做Streaming Ingestion就是在硬造轮子没有任何意义。反过来如果数据源就是一个活跃的MySQL业务库业务方要求下游特征更新在秒级那么批处理再怎么做也补不了这个缺口。再看下游消费方式。离线数仓Hive表、湖上Parquet分区表天然适合批量写入而OLAP里的实时聚合、Redis里的在线特征、Kafka to Kafka的实时清洗就必须走实时链路。最后一点是要确认“真实容忍度”。很多业务方说“要实时”翻译过来其实是“需要分钟级不想等天级”。如果你能通过把批处理调成每5分钟跑一次增量就能满足90%的需求那完全没必要上全套流计算。做Ingestion服务的人最怕的就是技术浪漫主义。3. 批处理摄取的服务设计调度、增量策略与SLA保证批处理摄取看起来简单无非拉数据、落表、收工。但一旦数据源多了问题就全冒出来了凌晨两点的上游接口超时、突然多了一个字段导致入库任务失败、某张表数据量突然翻倍把任务跑了六小时……批处理摄取的核心挑战不是“怎么拉”而是“怎么稳定地拉、可预期地完成”。3.1 调度框架怎么选调度层我推荐用DolphinScheduler或者Apache Airflow两者都是成熟的开源方案。Airflow生态大Python写DAG扩展性强适合技术能力强、愿意折腾的团队DolphinScheduler在易用性上更好自带分布式任务分发、补数操作、中文界面对中小企业更友好。调度选型要避免一个坑把调度做成一个大而全的脚本。有人习惯用一个Shell脚本串起所有Ingestion流程开始时爽后面每个加字段、重跑、失败告警都在里面打补丁最终变成一团乱麻。批处理链路必须拆成多个可独立重试的原子任务比如采集任务从上游源把数据拉到临时区校验任务比对上游行数、字段数、关键值sum值是否合理转换任务做必要的清洗和规范化加载任务写入目标表或目标文件验证任务抽样比对目标结果和源端预期每一步都有明确产物失败时只重跑失败的步骤不用整条链路上所有任务都重新执行一遍。3.2 增量抽取的几种思路批处理摄取不是每次把整张表拉一遍那是只有数据量小到可以忽略时才做的方案。真正常用的是增量抽取但各种增量方式各有适用边界。时间戳增量业务表里有updated_at之类的字段每次只取大于上次最大值的记录。这是最通用的方案但依赖上游在每次update时正确更新时间戳有些业务系统只在数据变化时更新部分字段或者只在应用层修改时间容易出现漏数据。自增ID增量每次取大于上次最大ID的记录。适用于只追加、几乎不改历史的表比如日志表、流水表。缺点是数据一旦修改不会重新进入摄取链路。Binlog/CDC增量通过解析数据库日志拿到变更记录。这其实已经接近实时的范畴了适合需要秒级或分钟级的数据后面实时摄取部分我会细说。全量分区覆盖按天或者按小时拉全量但写入内部系统时按时间分区覆盖只保留最新分区。数据量小、上游没时间戳时可用。实际设计时我建议对增量状态本身做持久化放到元数据库里记录每个数据源每次采样的起始偏移、成功状态、完成时间。这样便于补数和问题追溯也方便跨团队协作。3.3 SLA怎么算出来做批处理摄取一定要有明确的SLA定义否则出了问题没办法界定责任。我的习惯是给每个接入任务设置一个“最晚可用时间”比如上游凌晨1点出文件那么最晚早上7点必须完成内部落地给下游留出跑数时间。要保证SLA就得提前算清楚两个数字任务的基线耗时和可容忍的延迟缓冲。基线耗时通过历史运行数据的P95即95%的任务在这个时间内能跑完确定缓冲建议留30%到50%用于应对偶发的重试和网络抖动。另外批处理链路一定要设置“运行超时中断”机制。比如一个任务P95是30分钟你给它定的超时是60分钟超过就自动kill并告警。否则某个任务卡住时会一直占着资源拖慢整批任务最终SLA全部崩盘。4. 实时摄取的核心机制位移、状态与精确一次实时摄取听起来很高级但本质是要处理三个问题数据到了放哪里、从哪读、怎么保证不丢不重不乱。4.1 数据先进缓冲层再谈处理做实时摄取第一个建议就是不要把流数据直接打到最终的OLAP系统或业务系统。必须有一个高性能的缓冲层开路实践中默认是Kafka。Kafka作为缓冲有几个好处一是削峰填谷处理端可以按自己的节奏消费二是多消费者机制一份数据可以被离线、实时、告警等多个下游独立消费三是一段时间内数据依然可以回溯出了问题可以重新消费。上游数据进Kafka的方式要视数据源而定日志埋点、服务调用链数据直接用Filebeat/Fluentd采集到KafkaMySQL/PostgreSQL变更数据用Debezium Kafka Connect把binlog转成消息流第三方API数据通过HTTP Gateway服务转发到Kafka原有消息队列RocketMQ、RabbitMQ的存量数据写一个连接器消费后转投到Kafka4.2 Kafka的位移提交策略消费Kafka时最容易犯的错是“先提交位移再处理数据”。如果处理过程中宕机这批数据实际没有消费完成但位移已经提交造成数据丢失。反过来“先处理再提交”又有可能因为处理成功后、提交位移前宕机导致重启后重复消费。这就是经典的at-least-once和at-most-once取舍。实时摄取场景下我几乎都会选择at-least-once加下游幂等来兜底而不是耗费巨大精力去追求exactly-once。因为对绝大部分数据目标系统来说在目标端建立唯一键、靠upsert覆盖写入比在流处理引擎层靠分布式事务保证精确一次要简单可靠得多。举个例子消费订单消息后写入ClickHouse可以以order_id作为去重键用ReplacingMergeTree引擎或者INSERT ... ON DUPLICATE KEY UPDATE方式写入。即使重启后同一条订单被消费了两遍最终表里也只会有一条正确记录。这套方案简单、有效不需要引入额外的流处理引擎状态。4.3 Flink的状态机制与checkpoint如果是用Flink做实时摄取后的处理比如清洗、维表关联、聚合那就必须理解Flink的checkpoint机制。Flink的exactly-once依赖的是分布式快照——定期把各算子的状态和Kafka的消费位移一起持久化到外部存储比如HDFS或S3。checkpoint是一定要开的而且要仔细配置几个参数# 每个checkpoint之间的最小间隔防止频繁做快照拖垮吞吐 execution.checkpointing.min-pause30s # 一次checkpoint的超时时间超过即认为失败 execution.checkpointing.timeout10min # 允许的连续失败次数 execution.checkpointing.tolerable-failed-checkpoints3 # 存储到远端文件系统 state.checkpoints.dir: hdfs://nameservice/flink/checkpoints我见过不少团队不开checkpoint理由是“处理逻辑简单就清洗一下转发下游”。一旦Flink任务因为某个脏数据反序列化失败而挂掉重启后如果开着checkpoint可以从最近一次快照恢复消费位点回滚数据不会断也不会丢没有checkpoint的话要么从Kafka最早的位点重读导致大量重复要么从最新位点读导致窗口期数据直接丢失——两个选项都很痛。4.4 背压和窗口的取舍实时链路里如果某个下游处理过慢Flink会自动背压到Kafka消费者降低消费速率这是保护机制。好多人看到背压就紧张其实要分情况。短期瞬时背压没问题系统会自动消化如果长期背压且任务延迟持续拉长那就要查是下游写入瓶颈还是状态过大导致GC频繁。窗口设计上做实时摄取时尽量少用大窗口。大窗口会产生大量状态而且窗口计算的延迟和数据语义上的乱序修正都会变得复杂。如果场景允许优先用“事件驱动状态累加”代替“滚动窗口”。比如要统计近5分钟的订单金额与其用Flink的滑动窗口不如每来一条消息就累加到状态里再定时输出状态快照这样下游模型更简单问题也好定位。5. 服务化设计与数据质量让Ingestion服务真正好用很多团队把Ingestion做成一堆分散的脚本今天A同学加个Python采集脚本明天B同学加个Sqoop任务后天C同学写个Flink jar包。时间一长没人说得清整个平台究竟接了多少数据源、每个数据源的数据质量怎么样、出问题该找谁。这也是我认为Ingestion服务不能只是“技术能力”而是必须做成一个“产品”的原因。5.1 接入管理平台化的价值成熟的Ingestion服务应该有一个统一的接入管理界面把每个数据源做成一个配置化实例。接入时只需要填写数据源类型、连接信息、增量策略、目标表名、调度周期、告警接收人剩下的事情由平台自动生成采集任务、注册schema、配置监控。这种配置化管理最大的好处是可审计、可复用。每个接入实例都有清晰的所有者和负责人数据从哪来、要到哪去、是什么频率、是否存在问题一览无余。对于中大型团队这一步早晚要做晚做不如早做。5.2 Schema演化如何处理外部系统变数最大的就是数据结构字段会加、会删除、类型会变化。Ingestion服务处理schema变化我的原则是“避免推倒重来尽量向前兼容”。具体做法是在摄取层保存一份“最新schema”和“历史schema”历史数据按旧schema解释新数据按新schema写入目标。落地到Hive/Iceberg这类表结构时新加字段通过ALTER TABLE加列并给该列填充默认值对于删除的字段不要立刻物理删除保留一段时间防止下游还在用。关键的一点是schema变更要有通知机制一定要把变更告警发给数据接入的负责人和下游使用方。我见过太多线上事故根因就是上游偷偷加了个字段Ingestion服务自动兼容了但下游的ETL任务按旧字段取数取到的值全变成了NULL半夜跑挂了好几个报表任务。5.3 数据质量校验不能只依赖下游做Ingestion服务数据质量问题要在入口处拦截而不是等到了数仓里再由下游做数据质量规则发现。入口处至少要做这几层校验行数校验每次同步完成后对比源系统行数与目标表新增行数抽样校验对关键字段进行非空率、枚举值分布、最大最小值范围的抽查主键唯一性校验防止源端脏数据导致目标表重复业务规则校验比如金额不为负、日期字段格式合法、状态枚举值在允许范围内校验不通过时要能自动阻断并把数据放入“脏数据暂存区”同时给负责人发告警。这里要提个醒阻断链路和告警一定不能耦合。如果谁知道数据有问题结果告警网关也挂了整个部门就能安安静静地用脏数据跑一天这种教训太深刻了。5.4 可观测性建设告警不是越多越好Ingestion的监控指标最核心的是两个任务的时效性和数据准确率。时效性用“端到端延迟”来度量即从数据在源端产生到进入内部系统可查询的间隔数据准确率则靠前面说的质量校验规则来度量。告警方面宁可配置少而精准的告警也不要刷屏式告警。我的建议是最多配置三类告警任务失败或连续重试失败端到端延迟超过SLA的告警阈值数据质量校验的严重异常比如行数波动超过50%不要每条数据都设置告警。Ingestion链路偶尔出现一条坏数据很正常如果每次都半夜打call用不了两周团队成员对告警就会疲劳真正出大事时反而没人理。6. 真实链路里的几个大坑全是经验换来的这部分算是我用加班和线上事故堆出来的经验单独拎出来写每条都值得新做Ingestion服务的团队仔细对照。6.1 源端删除的数据增量同步永远看不见最坑的一种情况业务方在MySQL里订正数据删掉了昨天误插入的一批错误订单。你的批处理增量任务只看新增和更新删除操作不会体现在updated_at的变化里于是内部数据系统里永远是那些已经被删除的错误数据。解决这个问题没有银弹。可选方案包括源端不物理删除改成逻辑删除is_deleted1所有增量链路自动感知定期做全量对账用源端总行数、特定维度sum值和内部系统比对发现差异再触发订正如果实在只有物理删除那就只能依赖上游业务系统配合或者接受一定的数据不一致做Ingestion服务一定要在接入前就确认好源端删除策略不要等上线了再问。6.2 类型隐式转换带来的“数据漂移”上游字段是DECIMAL(10,2)内部系统建表时写成了DOUBLE看起来好像没什么问题但当某个字段的精度超过10位时DOUBLE的浮点误差就会显现。再比如上游是BIGINT的毫秒时间戳内部系统存的是DATETIMEETL转换时如果按秒解析所有时间都会偏到1970年。我的建议是Ingestion服务里对类型映射表做严格校验建表时自动根据源端类型选择最合适的内部类型并且对于浮点数、时间戳、金额这类敏感字段禁止自动推断。宁可建表时人工确认一次也不要上线后被数据的“神秘偏移”折磨。6.3 批量任务大量产出小文件批处理摄取如果处理不当很容易在目标存储上产生大量小文件。比如每5分钟同步一次MySQL的表数据到Hive每次都直接往目标分区里写新文件没做文件的合并/压缩一天下来这个表可能产生上千个小文件下游跑Spark SQL时光扫描元数据就要花掉70%的时间。解决方案是在摄入落地时先写入临时文件/临时分区在分区关闭前做一次文件合并和大小重整比如目标单文件控制在128MB256MB左右再原子性地把数据LOAD到正式分区。这样既保证了数据可见性也保证了文件布局的健康。6.4 实时链路的“顺序问题”比想象中更严重做实时摄取时很多人觉得把Kafka的partition设成1就能保证全局有序但这样牺牲了吞吐大部分场景下是得不偿失的。正确做法是保留多partition在写Kafka时按业务主键做partition key保证同一主键的变更一定进同一个partition从而保证单key的有序性。比如订单状态的流转消息以order_id作为key那么同一个订单的“创建、支付、发货、完成”事件一定被同一个消费者线程按序处理。这一步必须在Ingestion最开始就做好等数据已经分散到各个partition后再想保证有序基本只能重新消费重放。6.5 实时任务重启后的“回放风暴”流处理任务重启后状态恢复需要从checkpoint读取快照然后从Kafka位移处回放一段时间的数据。如果checkpoint过大或者下游被回放的数据量冲击很容易出现下游写库超时、连接池爆掉这些“二次灾害”。这里有两个实用手段尽量精细化checkpoint粒度不要把无关的状态都塞进同一个checkpoint缩短恢复时间在下游存储侧加写入限流或缓存批量写入机制比如写入批大小控制在5001000条一批防止瞬时流量打爆目标端实时链路和批处理链路不一样它没法通过“重跑一次”来恢复。所以任何操作都得多想一步“如果挂了恢复路径是什么”。7. 最后一个配置建议如果只让我给Ingestion服务加一个平台能力我会选“一键补数”。无论是批处理链路因为上游故障缺失了一段时间的数据还是实时链路因为bug导致某几个小时的数据质量异常只要历史数据在Kafka或源端还能拿到就能指定时间范围重新跑一遍摄取数据就会按照规则重新进入内部系统。这个能力需要Ingestion服务自始就把“幂等写入”作为硬性要求。也正是因为有了它数据接入的负责人才能在面对各类事故时底气足一点——先恢复增量链路再慢慢补历史数据而不是急得像热锅上的蚂蚁。做Ingestion服务不追求炫技追求的是把每个环节最朴素的事做到位。链路稳了下游怎么做都能心里有底。
