做大数据开发这些年接手的项目里“数据采集平台”是出现频率最高的一个。很多人一听这个名字觉得这不就是把数据从各个系统搬到数仓里嘛有什么好做的。但实际上一个真正能稳定跑下去的采集平台牵扯的细节远比想象得多——数据格式统一、延迟控制、增量与全量策略、分区管理、异常恢复哪一环没处理好后面全都要还债。这篇文章就围绕我最近完成的“基于Spark的数据采集平台”这个项目把从架构设计到核心实现、再到稳定性调优的完整思路梳理一遍给正在搭采集链路的朋友做个参考。整个平台的核心定位是用Spark作为统一的处理引擎把分散在各业务系统中的数据以可靠的方式抽取、解析、清洗后落入数仓目标表。适合数据团队规模不大、但又想建立标准化采集能力的场景。不管你是刚接触Spark还是已经在做数仓建设这篇文章都会有一些能直接用的东西。1. 项目定位与整体设计思路1.1 这个平台到底解决了什么问题先说一个容易被误解的点Spark本身不直接“采集”数据。它不像Flume那样监听日志文件也不像Canal那样直接挂在数据库上监听binlog。Spark在采集链路里担任的角色是“加工中心”——数据源通过采集客户端或者消息通道进入平台之后由Spark负责做统一的格式解析、清洗、转换再写入目标存储。换句话说采集平台是“采”和“存”之间的那一道关键工序。很多团队会有这样的经历业务方要看一份数据报表数仓同学去对接业务系统的数据库直接写个SQL抽数据然后每天定时跑一遍。刚开始数据量小还好等数据量上来、业务表越来越多就会出现几个问题——第一每张表一套抽取逻辑口径不统一第二源端直接执行全量查询压力大到影响业务系统第三缺失数据校验机制抽坏了谁都不知道。这个项目正是冲着这些问题来的用统一平台管理采集任务用Spark的分布式计算能力做数据加工把“能跑通”变成“稳定地跑通”。平台对标的典型场景是企业内部有多个业务系统数据源包括MySQL/Oracle这类关系型数据库、后端服务日志、IoT设备上报数据以及第三方接口推送的JSON数据。这些数据需要流入HDFS和数仓为后续的报表统计、用户画像、算法训练提供基础数据。适合了这个场景整套设计自然就有了明确约束。1.2 为什么选Spark作为核心引擎说到引擎选型目前主流的方案无外乎Spark和Flink两种。这两种我都做过实际项目说下取舍逻辑。Flink强在秒级延迟和精确一次性语义的流处理如果业务场景是实时风控、实时推荐这类对延迟极其敏感的应用Flink是更合适的选择。但数据采集平台的核心任务并不是“实时计算”而是“可靠搬运和整理”。在这个场景下我对延迟的要求通常是分钟级——比如新写入的数据5分钟后能查到就已经满足大多数报表需求了。Spark Structured Streaming的微批模式刚好覆盖这个区间又能直接复用Spark SQL的丰富API开发效率高出一大截。还有一个很现实的原因团队技术栈的统一。数据采集完总要去跑批处理任务如果一套平台用Spark、另一套平台用Flink运维成本和人员学习成本都会成倍增加。而Spark在批处理领域依然是事实标准选择Spark等于把“采集加工”和“批量ETL”两条链路统一到一个引擎上Hive Metastore里的表结构、以及Parquet这类列式存储格式都可以直接复用。当然选择Spark也意味着接受它的短板实时性上限是秒级到分钟级微批Structured Streaming在状态管理上不如Flink成熟。但从数据采集平台的体量来看这些短板不构成瓶颈。作为平台选型匹配需求比追逐时髦更重要。1.3 整体流程从数据源到目标存储平台的完整数据链路是这么走的数据源层各业务系统的关系型数据库、日志文件、接口推送数据采集通道层数据库用Sqoop或Canal同步日志用Filebeat采集后送入Kafka接口数据直接以JSON格式发送到Kafka缓冲层Kafka作为数据堆积的缓冲池削峰填谷避免业务高峰期直接压垮写入端处理层Spark批处理任务处理全量数据Spark Structured Streaming处理增量实时数据存储层处理后的数据写入HDFS数仓分层表按时间和业务分区组织应用层下游的数仓任务直接读取稳定、高质量的数据链条看着不复杂但每一层都有文章。比如Kafka的topic如何划分、消息格式怎么定、Spark消费Kafka的消费速率如何控制、HDFS上的目录命名如何规范这些细节决定了平台能不能健康运行。2. 平台架构与核心功能模块2.1 架构设计分层清晰职责单一我做的这套平台整体分为五个层级接入层。负责统一收口来自不同源的数据。关系型数据库的增量数据通过Canal解析binlog输出JSON格式消息到Kafka日志数据通过Filebeat采集写入Kafka业务系统主动推送的数据则直接调用统一的Kafka生产接口。这个设计避免了“一道数据一条链路”的混乱局面——不管数据源是谁到最后都汇入Kafka的专用topic。解析层。Spark在这里承担核心加工职责。流式任务消费Kafka中的原始JSON消息完成格式校验、字段名规范化、数据类型转换、以及业务规则过滤。用户画像和标签主题的数据还会在这一层完成简单的维表关联。存储层。处理后的数据写入HDFS的数仓分层目录。基础数据落在ODS层按业务事件时间做分区轻度汇总数据落在DWD层清洗后按主题domain组织。统一使用Parquet列式存储配合Snappy压缩既节省空间又保证查询性能。调度层。任务调度依托业务方的调度系统采集任务以Shell脚本方式提交设置周期间隔依赖。整个平台在凌晨跑全量任务全天滚动跑增量任务。元数据与监控层。采集任务记录数据文件的行数、大小、跑批开始时间和结束时间写入监控表。每天自动对账如果某张表的产出行数和源端预估行数偏差超过阈值触发告警。这里有一个很核心的设计思路把“生产原始数据”和“加工标准数据”拆分开。Kafka里存储的是最原始的JSON消息保留原始信息Spark处理阶段不允许使用任何“临时补丁”式的规则所有逻辑变动都要走代码评审。这样做虽然看起来死板却在长期维护上省下了大功夫——不会出现某天某个同事在任务里临时加了一个filter三个月后谁都不知道数据为什么少了。2.2 核心模块设计全量同步模块、增量同步模块、质量校验模块围绕数据加工逻辑平台拆成了三个核心模块。全量同步模块处理历史数据回补和维表数据定时拉全量。典型场景是用户维度表业务上需要每12小时拉一次全量保持维表最新。实现上就是使用Spark JDBC读取源表按主键哈希分片并行读取减少源库压力然后直接覆写目标分区。增量同步模块对接Kafka中的binlog事件通过Spark Structured Streaming消费。这个模块是全平台的“心脏”因为增量数据是实时涌入的消费落后、解析失败、乱序到达这些问题都会在这里被放大。实现对binlog消息做了三层保护格式校验错误则写入死信队列业务规则无法解析则记录原始日志可容忍的延迟乱序数据依靠watermark机制来处理。质量校验模块针对关键结果表每次任务结束后做行数波动的区间校验。具体来说计算前7天同表行数的均值和标准差当前行数落点超过均线±3倍标准差任务状态置为异常并人工确认。这个模块看起来简单却在数次源端数据错乱事件中提前报警及时止损。事后复盘发现都是源库误删或接口改字段的问题。2.3 数据库选型与存储分区的关键设计存储层是最容易被忽略、却最影响后期维护的部分。我的经验是在搭建第一天就把目录规格定清楚表名不清晰后期再改成本极高。HDFS路径规范数据统一放在/warehouse/{层名}/{主题域名}/{表名}下在这套平台里就是/warehouse/ods/trade_orders、/warehouse/dwd/user_profile这样的结构。分区统一使用pt字段格式为yyyyMMdd小时级数据以hour字段为二级分区。文件格式统一ParquetSnappy。原因有三点Parquet的列式存储对统计查询友好Snappy压缩比合理且解压速度快Spark读写Parquet的优化做得最成熟。早期项目里我见过用CSV裸存的查询慢还在小文件问题上吃尽苦头换掉之后性能相差一个量级。库表格式Hive建表统一采用事务表或者外部表数据文件和分区路径严格匹配。每一张采集结果表都记录在元数据登记表里字段类型、业务负责人、更新频率、数据来源系统信息缺一不可。采集平台看似是“往下游投喂数据”实际上它也是数仓平台的“上游依赖”信息不全最后受苦的是下游。3. 核心任务实现与技术细节3.1 全量采集任务的代码实现全量采集功能听起来简单——读源表、写目标表——但直接这么干会踩坑。核心问题在于全表扫描对源库的压力以及大数据量写入目标时小文件失控。下面是我在平台中用的方式。// 全量采集任务核心代码Scala val sourceTable source_db.t_orders val targetPath /warehouse/ods/trade_orders/pt bizDate val jdbcDF spark.read .format(jdbc) .option(url, sourceDbUrl) .option(user, sourceDbUser) .option(password, sourceDbPassword) .option(dbtable, sourceTable) .option(numPartitions, 16) // 并行读取分区数 .option(partitionColumn, order_id) // 分片字段需要是数字型 .option(lowerBound, minId) .option(upperBound, maxId) .option(fetchSize, 5000) .load() // 分区字段过滤只提取当天新增数据减少写入压力 val filteredDF jdbcDF.where(col(create_time) startTime) filteredDF .repartition(8, col(order_id)) // 控制输出文件数量避免大量小文件 .write .mode(overwrite) .format(parquet) .save(targetPath)这里有两个关键参数要细说。第一个是numPartitions和partitionColumn的配合。partitionColumn指定的字段必须是一张表的唯一索引或明确有索引的数字列让Spark拉取数据时按该字段的范围拆分成多个并行查询比如WHERE order_id BETWEEN 1 AND 10000和WHERE order_id BETWEEN 10001 AND 20000。JDBC并行分区是采集平台中最重要的调优点分区数设少了读取慢设多了源库直接被榨干。经验值是在源库性能可接受的前提下分区数为源表数据总量的十万分之一到五十万分之一单分区数据量控制在 500MB 左右。第二个是repartition(8, col(order_id))。如果不做这步Spark 从JDBC读取后文件数可能等于分区数导致输出大量小文件。小文件问题在HDFS上是个长期症状——NameNode 内存被大量无效元数据占满下游读表时启动的task数量剧增整体性能恶化。所以全量任务写之前我会刻意控制目标文件数按照单文件 128MB 的期望大小做重分区。如果每张表数据量差异很大分区数需要在一段实践后固定下来。全量任务在平台中以Shell脚本方式封装核心命令如下spark-submit \ --class com.example.platform.dataloader.FullLoadRunner \ --master yarn \ --deploy-mode cluster \ --queue root.etl \ --driver-memory 4g \ --num-executors 20 \ --executor-memory 8g \ --executor-cores 4 \ full-load-job.jar \ --source orders_db \ --table t_orders \ --biz-date 2024-01-01提交参数不是随意设置的。--queue root.etl指定了独立的资源池避免采集任务和日常分析任务抢占Yarn资源。executor-memory 8g和executor-cores 4的组合在多数场景下能较好地平衡并行度和内存溢出风险。内存再往上调不一定划算——单个Executor内存超过12G后GC停顿时间会明显增加。3.2 增量采集任务Structured Streaming消费Kafka增量数据链路是全天不间断运行的。增量模块用Spark Structured Streaming消费Kafka中由Canal写入的binlog信息。先看核心代码// 增量采集任务核心代码Scala val kafkaStream spark.readStream .format(kafka) .option(kafka.bootstrap.servers, kafkaBrokers) .option(subscribe, canal_order_binlog) .option(startingOffsets, latest) // 从最新偏移量开始避免重复消费 .option(maxOffsetsPerTrigger, 200000) // 每批最多消费的offset数控制吞吐量 .load() .selectExpr(CAST(value AS STRING) AS message) // 解析JSON并提取事件类型 val parseDF kafkaStream .select(from_json(col(message), binlogSchema).alias(data)) .select( col(data.table), col(data.event_type), col(data.order_id), col(data.create_time) ) .filter(col(event_type).isin(INSERT, UPDATE, DELETE)) // 写入HDFS按事件日期做分区 val query parseDF .writeStream .outputMode(append) .format(parquet) .partitionBy(dt) .option(path, /warehouse/ods/trade_orders) .option(checkpointLocation, /spark-checkpoint/ods_trade_orders) .trigger(Trigger.ProcessingTime(2 minutes)) .start() query.awaitTermination()这段代码整体逻辑清晰但有两个必须关注的参数。maxOffsetsPerTrigger管的是消费者吞吐的上限。它希望每两分钟处理多少数据取决于下游HDFS写入能力和上游Kafka堆积情况。设得太小增量数据实时性差设得太大一次性刷入大量数据下游任务调度会出现毛刺。我在平台上常用的设置是每批200000条对应按平均消息体1KB计算也就是每批200MB的数据量两分钟一批完全能消化。checkpointLocation参数的指定位置是流式应用的生命线。如果任务意外终止Structured Streaming从checkpoint记录中恢复未完成批次偏移量保证不丢不重。强调“不重”是基于At-Least-Once在配合幂等输出时实现的同时checkpointLocation路径要做好版本管理路径一旦变化等同于新作业会从startingOffsets重新消费可能造成数据重复。这一点在我早期的一次误操作中让人印象深刻——手滑改了checkpoint路径导致当天数据翻倍下游汇总任务算了两遍结果。后来专门写了一条规则checkpoint路径固定永不修改除非整体重建。3.3 数据切分与时间语义事件时间与处理时间的坑数据采集平台很容易被时间字段搞出问题。上游业务系统数据库里经常存在数据到达顺序错乱的情况例如某条订单是 12:00:00 创建的但binlog直到 12:05:00 才被采集或者移动端离线缓存的数据在夜间集中补报。Structured Streaming里引入了两个时间概念event time业务发生时间和processing time任务处理时间。增量采集平台里主要依据事件时间进行分区配合watermark处理迟到的数据val withWatermark parseDF .withColumn(event_time, to_timestamp(col(create_time), yyyy-MM-dd HH:mm:ss)) .withWatermark(event_time, 10 minutes) // 允许10分钟延迟超出丢弃或走修正逻辑 .groupBy(window(col(event_time), 5 minutes)) .count()这里究竟该设多长watermark窗口取决于业务容忍度。平台中大多数业务表设置了10分钟。如果有些表来源网络波动大或者IoT设备信号不稳定会设置到30分钟。watermark设置太短会有大量数据迟到被弃太长则拉低“增量”的实时语义。二者需要根据业务容忍度进行平衡。另外注意流式作业落地数据时如果数据迟到了但已经落入前一小时分区我不会去向已经写出的HDFS分区做实时回改而是记入“迟到队列”第二天凌晨由定时任务做补数和合并。很多线上事故都源于试图修改已落盘的历史分区并发写同一目录会发生Task冲突。4. 性能调优与稳定性保障4.1 Executor调优兼顾并行与资源Spark作业的参数调优是采集平台稳定运行的关键。以我线上最常用的增量作业为例Yarn资源分配如下参数配置值说明num-executors30并发执行的Executor数量executor-memory6g每个Executor内存executor-cores3每个Executor使用的CPU核数spark.sql.shuffle.partitions200Shuffle分区数量spark.sql.adaptive.enabledtrue开启自适应查询执行避免小任务浪费资源spark.sql.adaptive.coalescePartitions.enabledtrue自动合并小分区spark.streaming.kafka.maxRatePerPartition10000每分区每秒最大消费速率Executor数量的设置核心逻辑受限于消耗的资源队列和源端Kafka的QPS上限。30个Executor、每Executor 3个并发核高峰期能产生的最大并发度为90。如果Topic分区数为36那么理论上同时最多处理36个分区的消费再多数量的Executor也只是空闲等待。有一个在高并发场景下值得注意的参数是maxRatePerPartition。这个参数硬性限定每分区每秒消费的速率避免消费速度超过下游写入效率导致“获取数据-处理数据”的颠簸。把消费速率控制在10000条每秒叠加maxOffsetsPerTrigger双重限制可以让流量呈平滑的波状让下游存储有持续稳定的写入压力。4.2 数据倾斜与防重防丢的实战措施数据倾斜是Spark任务最常见的“杀手”。在数据场景里业务的热门订单数量可能远超普通订单。如果按order_id哈希分区一个热点用户ID产生的大量订单会全部堆积到一个分区。表现就是某个Executor跑了一小时还没结束其他Executor早就闲置。实战中对订单数据应对数据倾斜有几个手段。加盐给order_id拼接一个随机后缀打散热点Key显示洗完后在最终聚合时去掉前缀。这个方案实现简单负面影响是需要多做一次聚合CPU开销增加10%-15%。在业务能容忍的时间窗口内值得。过滤热点先按照频率识别出top10的热点Key单独计算后union回去。另一个不得不提的是“数据重复”。在处理binlog时如果Canal消费端重启或者Spark的某个任务挂掉重试同一事件有可能被处理两次。平台层面应对的策略是“目标端幂等写入”全量表直接按分区覆盖写天然幂等增量表的分区按“append”模式写入但额外增加一个event_id字段——该字段由源事件主键变更时间生成下游在使用时用ROW_NUMBER()按用户维度去重挑出最新一条。这样即使在处理链路里出现重复最终表里也能保证单条事件有且仅有一条生效。这套方案用一个小小的代价换来了长期稳定代价是每张增量表要多做一次event_id维度的按事件最新记录取数的操作。报表读取再也不用担心数据翻倍问题。4.3 常见问题排查指南采集平台上线后最常见的故障集中在以下几类。我给每个都做了一个速查记录一、任务提交完成但HDFS上没有数据文件。排查步骤先用yarn logs -applicationId xxx看Driver日志。大概率问题出在JDBC连接失败或者上游库权限变更。推荐在提交任务前先跑10行数据量的验证任务确认源库连通和数据格式正确再跑正式任务。少跑很多冤枉路。二、Structured Streaming任务一直报Kafka Offset Out of Range。原因通常是两种offset设置了earliest但checkpoint里保存的offset超过了Kafka里现存的最早offset消息陈旧被过期清理或者topic最近被扩容分区数变了。处理方式先确认checkpoint保存的offset是否正常如果期间有无任务运行但消息过期需要调整Kafka的retention.ms线上一般放宽到3天让业务有起码的容错窗口。三、增量数据偶发丢失。排查重点集中在Canal侧。Canal在捕捉binlog时如果网络断了一段恢复后可能导致一段binlog没有解析。平台在设计时让Canal同时保留一份原始增量文件到备份目录Spark任务启动时先做“数据缺口检测”——比较源主题表和存储层的最大事件时间超过阈值则自动触发一份增量回补任务。四、性能下降从十分钟降到半小时。最常见的元凶是HDFS小文件数量增加。在流式处理下每分钟输出一个小的parquet文件积少成多。处理方式是引入定期合并任务对数据量小的分区做rewrite合并后文件大小控制在128MB左右。保险的做法是让Spark在write前执行repartition(1)——不过要权衡文件少了对下游查询快但写入速度慢对实时链路有影响。平台的选择是保留合理的并发度定期做文件合并任务。我在这个项目搭建过程中最深的体会是稳定压倒一切。数据采集平台不像算法模型有炫酷的产出它的成就感来自于“上百个任务每天准时跑完没有任何一个表数据缺失”。要做到这一点靠的不是某一次的精妙代码而是把监控、报警、幂等、幂等重试这些“笨功夫”做到位。这套基于Spark的数据采集平台上线后接手运维的同事反馈说最大的改观就是“省心”——不用再整天追着任务问数据到哪了所有的状态都在监控页面上写清楚。如果你正在搭自己的采集平台建议先把全量、增量、质量校验三块基础打牢资源调优和特性优化可以等跑通后再迭代。 
长河君 · 2026/10/15 11:15:05 · 阅读 1.2k 基于Spark的数据采集平台：全量增量同步与架构调优实践
做大数据开发这些年接手的项目里“数据采集平台”是出现频率最高的一个。很多人一听这个名字觉得这不就是把数据从各个系统搬到数仓里嘛有什么好做的。但实际上一个真正能稳定跑下去的采集平台牵扯的细节远比想象得多——数据格式统一、延迟控制、增量与全量策略、分区管理、异常恢复哪一环没处理好后面全都要还债。这篇文章就围绕我最近完成的“基于Spark的数据采集平台”这个项目把从架构设计到核心实现、再到稳定性调优的完整思路梳理一遍给正在搭采集链路的朋友做个参考。整个平台的核心定位是用Spark作为统一的处理引擎把分散在各业务系统中的数据以可靠的方式抽取、解析、清洗后落入数仓目标表。适合数据团队规模不大、但又想建立标准化采集能力的场景。不管你是刚接触Spark还是已经在做数仓建设这篇文章都会有一些能直接用的东西。1. 项目定位与整体设计思路1.1 这个平台到底解决了什么问题先说一个容易被误解的点Spark本身不直接“采集”数据。它不像Flume那样监听日志文件也不像Canal那样直接挂在数据库上监听binlog。Spark在采集链路里担任的角色是“加工中心”——数据源通过采集客户端或者消息通道进入平台之后由Spark负责做统一的格式解析、清洗、转换再写入目标存储。换句话说采集平台是“采”和“存”之间的那一道关键工序。很多团队会有这样的经历业务方要看一份数据报表数仓同学去对接业务系统的数据库直接写个SQL抽数据然后每天定时跑一遍。刚开始数据量小还好等数据量上来、业务表越来越多就会出现几个问题——第一每张表一套抽取逻辑口径不统一第二源端直接执行全量查询压力大到影响业务系统第三缺失数据校验机制抽坏了谁都不知道。这个项目正是冲着这些问题来的用统一平台管理采集任务用Spark的分布式计算能力做数据加工把“能跑通”变成“稳定地跑通”。平台对标的典型场景是企业内部有多个业务系统数据源包括MySQL/Oracle这类关系型数据库、后端服务日志、IoT设备上报数据以及第三方接口推送的JSON数据。这些数据需要流入HDFS和数仓为后续的报表统计、用户画像、算法训练提供基础数据。适合了这个场景整套设计自然就有了明确约束。1.2 为什么选Spark作为核心引擎说到引擎选型目前主流的方案无外乎Spark和Flink两种。这两种我都做过实际项目说下取舍逻辑。Flink强在秒级延迟和精确一次性语义的流处理如果业务场景是实时风控、实时推荐这类对延迟极其敏感的应用Flink是更合适的选择。但数据采集平台的核心任务并不是“实时计算”而是“可靠搬运和整理”。在这个场景下我对延迟的要求通常是分钟级——比如新写入的数据5分钟后能查到就已经满足大多数报表需求了。Spark Structured Streaming的微批模式刚好覆盖这个区间又能直接复用Spark SQL的丰富API开发效率高出一大截。还有一个很现实的原因团队技术栈的统一。数据采集完总要去跑批处理任务如果一套平台用Spark、另一套平台用Flink运维成本和人员学习成本都会成倍增加。而Spark在批处理领域依然是事实标准选择Spark等于把“采集加工”和“批量ETL”两条链路统一到一个引擎上Hive Metastore里的表结构、以及Parquet这类列式存储格式都可以直接复用。当然选择Spark也意味着接受它的短板实时性上限是秒级到分钟级微批Structured Streaming在状态管理上不如Flink成熟。但从数据采集平台的体量来看这些短板不构成瓶颈。作为平台选型匹配需求比追逐时髦更重要。1.3 整体流程从数据源到目标存储平台的完整数据链路是这么走的数据源层各业务系统的关系型数据库、日志文件、接口推送数据采集通道层数据库用Sqoop或Canal同步日志用Filebeat采集后送入Kafka接口数据直接以JSON格式发送到Kafka缓冲层Kafka作为数据堆积的缓冲池削峰填谷避免业务高峰期直接压垮写入端处理层Spark批处理任务处理全量数据Spark Structured Streaming处理增量实时数据存储层处理后的数据写入HDFS数仓分层表按时间和业务分区组织应用层下游的数仓任务直接读取稳定、高质量的数据链条看着不复杂但每一层都有文章。比如Kafka的topic如何划分、消息格式怎么定、Spark消费Kafka的消费速率如何控制、HDFS上的目录命名如何规范这些细节决定了平台能不能健康运行。2. 平台架构与核心功能模块2.1 架构设计分层清晰职责单一我做的这套平台整体分为五个层级接入层。负责统一收口来自不同源的数据。关系型数据库的增量数据通过Canal解析binlog输出JSON格式消息到Kafka日志数据通过Filebeat采集写入Kafka业务系统主动推送的数据则直接调用统一的Kafka生产接口。这个设计避免了“一道数据一条链路”的混乱局面——不管数据源是谁到最后都汇入Kafka的专用topic。解析层。Spark在这里承担核心加工职责。流式任务消费Kafka中的原始JSON消息完成格式校验、字段名规范化、数据类型转换、以及业务规则过滤。用户画像和标签主题的数据还会在这一层完成简单的维表关联。存储层。处理后的数据写入HDFS的数仓分层目录。基础数据落在ODS层按业务事件时间做分区轻度汇总数据落在DWD层清洗后按主题domain组织。统一使用Parquet列式存储配合Snappy压缩既节省空间又保证查询性能。调度层。任务调度依托业务方的调度系统采集任务以Shell脚本方式提交设置周期间隔依赖。整个平台在凌晨跑全量任务全天滚动跑增量任务。元数据与监控层。采集任务记录数据文件的行数、大小、跑批开始时间和结束时间写入监控表。每天自动对账如果某张表的产出行数和源端预估行数偏差超过阈值触发告警。这里有一个很核心的设计思路把“生产原始数据”和“加工标准数据”拆分开。Kafka里存储的是最原始的JSON消息保留原始信息Spark处理阶段不允许使用任何“临时补丁”式的规则所有逻辑变动都要走代码评审。这样做虽然看起来死板却在长期维护上省下了大功夫——不会出现某天某个同事在任务里临时加了一个filter三个月后谁都不知道数据为什么少了。2.2 核心模块设计全量同步模块、增量同步模块、质量校验模块围绕数据加工逻辑平台拆成了三个核心模块。全量同步模块处理历史数据回补和维表数据定时拉全量。典型场景是用户维度表业务上需要每12小时拉一次全量保持维表最新。实现上就是使用Spark JDBC读取源表按主键哈希分片并行读取减少源库压力然后直接覆写目标分区。增量同步模块对接Kafka中的binlog事件通过Spark Structured Streaming消费。这个模块是全平台的“心脏”因为增量数据是实时涌入的消费落后、解析失败、乱序到达这些问题都会在这里被放大。实现对binlog消息做了三层保护格式校验错误则写入死信队列业务规则无法解析则记录原始日志可容忍的延迟乱序数据依靠watermark机制来处理。质量校验模块针对关键结果表每次任务结束后做行数波动的区间校验。具体来说计算前7天同表行数的均值和标准差当前行数落点超过均线±3倍标准差任务状态置为异常并人工确认。这个模块看起来简单却在数次源端数据错乱事件中提前报警及时止损。事后复盘发现都是源库误删或接口改字段的问题。2.3 数据库选型与存储分区的关键设计存储层是最容易被忽略、却最影响后期维护的部分。我的经验是在搭建第一天就把目录规格定清楚表名不清晰后期再改成本极高。HDFS路径规范数据统一放在/warehouse/{层名}/{主题域名}/{表名}下在这套平台里就是/warehouse/ods/trade_orders、/warehouse/dwd/user_profile这样的结构。分区统一使用pt字段格式为yyyyMMdd小时级数据以hour字段为二级分区。文件格式统一ParquetSnappy。原因有三点Parquet的列式存储对统计查询友好Snappy压缩比合理且解压速度快Spark读写Parquet的优化做得最成熟。早期项目里我见过用CSV裸存的查询慢还在小文件问题上吃尽苦头换掉之后性能相差一个量级。库表格式Hive建表统一采用事务表或者外部表数据文件和分区路径严格匹配。每一张采集结果表都记录在元数据登记表里字段类型、业务负责人、更新频率、数据来源系统信息缺一不可。采集平台看似是“往下游投喂数据”实际上它也是数仓平台的“上游依赖”信息不全最后受苦的是下游。3. 核心任务实现与技术细节3.1 全量采集任务的代码实现全量采集功能听起来简单——读源表、写目标表——但直接这么干会踩坑。核心问题在于全表扫描对源库的压力以及大数据量写入目标时小文件失控。下面是我在平台中用的方式。// 全量采集任务核心代码Scala val sourceTable source_db.t_orders val targetPath /warehouse/ods/trade_orders/pt bizDate val jdbcDF spark.read .format(jdbc) .option(url, sourceDbUrl) .option(user, sourceDbUser) .option(password, sourceDbPassword) .option(dbtable, sourceTable) .option(numPartitions, 16) // 并行读取分区数 .option(partitionColumn, order_id) // 分片字段需要是数字型 .option(lowerBound, minId) .option(upperBound, maxId) .option(fetchSize, 5000) .load() // 分区字段过滤只提取当天新增数据减少写入压力 val filteredDF jdbcDF.where(col(create_time) startTime) filteredDF .repartition(8, col(order_id)) // 控制输出文件数量避免大量小文件 .write .mode(overwrite) .format(parquet) .save(targetPath)这里有两个关键参数要细说。第一个是numPartitions和partitionColumn的配合。partitionColumn指定的字段必须是一张表的唯一索引或明确有索引的数字列让Spark拉取数据时按该字段的范围拆分成多个并行查询比如WHERE order_id BETWEEN 1 AND 10000和WHERE order_id BETWEEN 10001 AND 20000。JDBC并行分区是采集平台中最重要的调优点分区数设少了读取慢设多了源库直接被榨干。经验值是在源库性能可接受的前提下分区数为源表数据总量的十万分之一到五十万分之一单分区数据量控制在 500MB 左右。第二个是repartition(8, col(order_id))。如果不做这步Spark 从JDBC读取后文件数可能等于分区数导致输出大量小文件。小文件问题在HDFS上是个长期症状——NameNode 内存被大量无效元数据占满下游读表时启动的task数量剧增整体性能恶化。所以全量任务写之前我会刻意控制目标文件数按照单文件 128MB 的期望大小做重分区。如果每张表数据量差异很大分区数需要在一段实践后固定下来。全量任务在平台中以Shell脚本方式封装核心命令如下spark-submit \ --class com.example.platform.dataloader.FullLoadRunner \ --master yarn \ --deploy-mode cluster \ --queue root.etl \ --driver-memory 4g \ --num-executors 20 \ --executor-memory 8g \ --executor-cores 4 \ full-load-job.jar \ --source orders_db \ --table t_orders \ --biz-date 2024-01-01提交参数不是随意设置的。--queue root.etl指定了独立的资源池避免采集任务和日常分析任务抢占Yarn资源。executor-memory 8g和executor-cores 4的组合在多数场景下能较好地平衡并行度和内存溢出风险。内存再往上调不一定划算——单个Executor内存超过12G后GC停顿时间会明显增加。3.2 增量采集任务Structured Streaming消费Kafka增量数据链路是全天不间断运行的。增量模块用Spark Structured Streaming消费Kafka中由Canal写入的binlog信息。先看核心代码// 增量采集任务核心代码Scala val kafkaStream spark.readStream .format(kafka) .option(kafka.bootstrap.servers, kafkaBrokers) .option(subscribe, canal_order_binlog) .option(startingOffsets, latest) // 从最新偏移量开始避免重复消费 .option(maxOffsetsPerTrigger, 200000) // 每批最多消费的offset数控制吞吐量 .load() .selectExpr(CAST(value AS STRING) AS message) // 解析JSON并提取事件类型 val parseDF kafkaStream .select(from_json(col(message), binlogSchema).alias(data)) .select( col(data.table), col(data.event_type), col(data.order_id), col(data.create_time) ) .filter(col(event_type).isin(INSERT, UPDATE, DELETE)) // 写入HDFS按事件日期做分区 val query parseDF .writeStream .outputMode(append) .format(parquet) .partitionBy(dt) .option(path, /warehouse/ods/trade_orders) .option(checkpointLocation, /spark-checkpoint/ods_trade_orders) .trigger(Trigger.ProcessingTime(2 minutes)) .start() query.awaitTermination()这段代码整体逻辑清晰但有两个必须关注的参数。maxOffsetsPerTrigger管的是消费者吞吐的上限。它希望每两分钟处理多少数据取决于下游HDFS写入能力和上游Kafka堆积情况。设得太小增量数据实时性差设得太大一次性刷入大量数据下游任务调度会出现毛刺。我在平台上常用的设置是每批200000条对应按平均消息体1KB计算也就是每批200MB的数据量两分钟一批完全能消化。checkpointLocation参数的指定位置是流式应用的生命线。如果任务意外终止Structured Streaming从checkpoint记录中恢复未完成批次偏移量保证不丢不重。强调“不重”是基于At-Least-Once在配合幂等输出时实现的同时checkpointLocation路径要做好版本管理路径一旦变化等同于新作业会从startingOffsets重新消费可能造成数据重复。这一点在我早期的一次误操作中让人印象深刻——手滑改了checkpoint路径导致当天数据翻倍下游汇总任务算了两遍结果。后来专门写了一条规则checkpoint路径固定永不修改除非整体重建。3.3 数据切分与时间语义事件时间与处理时间的坑数据采集平台很容易被时间字段搞出问题。上游业务系统数据库里经常存在数据到达顺序错乱的情况例如某条订单是 12:00:00 创建的但binlog直到 12:05:00 才被采集或者移动端离线缓存的数据在夜间集中补报。Structured Streaming里引入了两个时间概念event time业务发生时间和processing time任务处理时间。增量采集平台里主要依据事件时间进行分区配合watermark处理迟到的数据val withWatermark parseDF .withColumn(event_time, to_timestamp(col(create_time), yyyy-MM-dd HH:mm:ss)) .withWatermark(event_time, 10 minutes) // 允许10分钟延迟超出丢弃或走修正逻辑 .groupBy(window(col(event_time), 5 minutes)) .count()这里究竟该设多长watermark窗口取决于业务容忍度。平台中大多数业务表设置了10分钟。如果有些表来源网络波动大或者IoT设备信号不稳定会设置到30分钟。watermark设置太短会有大量数据迟到被弃太长则拉低“增量”的实时语义。二者需要根据业务容忍度进行平衡。另外注意流式作业落地数据时如果数据迟到了但已经落入前一小时分区我不会去向已经写出的HDFS分区做实时回改而是记入“迟到队列”第二天凌晨由定时任务做补数和合并。很多线上事故都源于试图修改已落盘的历史分区并发写同一目录会发生Task冲突。4. 性能调优与稳定性保障4.1 Executor调优兼顾并行与资源Spark作业的参数调优是采集平台稳定运行的关键。以我线上最常用的增量作业为例Yarn资源分配如下参数配置值说明num-executors30并发执行的Executor数量executor-memory6g每个Executor内存executor-cores3每个Executor使用的CPU核数spark.sql.shuffle.partitions200Shuffle分区数量spark.sql.adaptive.enabledtrue开启自适应查询执行避免小任务浪费资源spark.sql.adaptive.coalescePartitions.enabledtrue自动合并小分区spark.streaming.kafka.maxRatePerPartition10000每分区每秒最大消费速率Executor数量的设置核心逻辑受限于消耗的资源队列和源端Kafka的QPS上限。30个Executor、每Executor 3个并发核高峰期能产生的最大并发度为90。如果Topic分区数为36那么理论上同时最多处理36个分区的消费再多数量的Executor也只是空闲等待。有一个在高并发场景下值得注意的参数是maxRatePerPartition。这个参数硬性限定每分区每秒消费的速率避免消费速度超过下游写入效率导致“获取数据-处理数据”的颠簸。把消费速率控制在10000条每秒叠加maxOffsetsPerTrigger双重限制可以让流量呈平滑的波状让下游存储有持续稳定的写入压力。4.2 数据倾斜与防重防丢的实战措施数据倾斜是Spark任务最常见的“杀手”。在数据场景里业务的热门订单数量可能远超普通订单。如果按order_id哈希分区一个热点用户ID产生的大量订单会全部堆积到一个分区。表现就是某个Executor跑了一小时还没结束其他Executor早就闲置。实战中对订单数据应对数据倾斜有几个手段。加盐给order_id拼接一个随机后缀打散热点Key显示洗完后在最终聚合时去掉前缀。这个方案实现简单负面影响是需要多做一次聚合CPU开销增加10%-15%。在业务能容忍的时间窗口内值得。过滤热点先按照频率识别出top10的热点Key单独计算后union回去。另一个不得不提的是“数据重复”。在处理binlog时如果Canal消费端重启或者Spark的某个任务挂掉重试同一事件有可能被处理两次。平台层面应对的策略是“目标端幂等写入”全量表直接按分区覆盖写天然幂等增量表的分区按“append”模式写入但额外增加一个event_id字段——该字段由源事件主键变更时间生成下游在使用时用ROW_NUMBER()按用户维度去重挑出最新一条。这样即使在处理链路里出现重复最终表里也能保证单条事件有且仅有一条生效。这套方案用一个小小的代价换来了长期稳定代价是每张增量表要多做一次event_id维度的按事件最新记录取数的操作。报表读取再也不用担心数据翻倍问题。4.3 常见问题排查指南采集平台上线后最常见的故障集中在以下几类。我给每个都做了一个速查记录一、任务提交完成但HDFS上没有数据文件。排查步骤先用yarn logs -applicationId xxx看Driver日志。大概率问题出在JDBC连接失败或者上游库权限变更。推荐在提交任务前先跑10行数据量的验证任务确认源库连通和数据格式正确再跑正式任务。少跑很多冤枉路。二、Structured Streaming任务一直报Kafka Offset Out of Range。原因通常是两种offset设置了earliest但checkpoint里保存的offset超过了Kafka里现存的最早offset消息陈旧被过期清理或者topic最近被扩容分区数变了。处理方式先确认checkpoint保存的offset是否正常如果期间有无任务运行但消息过期需要调整Kafka的retention.ms线上一般放宽到3天让业务有起码的容错窗口。三、增量数据偶发丢失。排查重点集中在Canal侧。Canal在捕捉binlog时如果网络断了一段恢复后可能导致一段binlog没有解析。平台在设计时让Canal同时保留一份原始增量文件到备份目录Spark任务启动时先做“数据缺口检测”——比较源主题表和存储层的最大事件时间超过阈值则自动触发一份增量回补任务。四、性能下降从十分钟降到半小时。最常见的元凶是HDFS小文件数量增加。在流式处理下每分钟输出一个小的parquet文件积少成多。处理方式是引入定期合并任务对数据量小的分区做rewrite合并后文件大小控制在128MB左右。保险的做法是让Spark在write前执行repartition(1)——不过要权衡文件少了对下游查询快但写入速度慢对实时链路有影响。平台的选择是保留合理的并发度定期做文件合并任务。我在这个项目搭建过程中最深的体会是稳定压倒一切。数据采集平台不像算法模型有炫酷的产出它的成就感来自于“上百个任务每天准时跑完没有任何一个表数据缺失”。要做到这一点靠的不是某一次的精妙代码而是把监控、报警、幂等、幂等重试这些“笨功夫”做到位。这套基于Spark的数据采集平台上线后接手运维的同事反馈说最大的改观就是“省心”——不用再整天追着任务问数据到哪了所有的状态都在监控页面上写清楚。如果你正在搭自己的采集平台建议先把全量、增量、质量校验三块基础打牢资源调优和特性优化可以等跑通后再迭代。