★ FEATURED ARTICLE 做数据同步好几年了从最早用定时脚本扫表到后来接CDCChange Data Capture变更数据捕获管道再到用Kafka把整个公司的实时数据链路串起来回头看看Kafka在大数据领域实时数据同步这个方向确实是绕不开的核心组件。很多人一提到大数据就想到Hadoop、Spark但真正在做实时数仓、数据湖、下游业务数据同步的时候Kafka承担的角色更像是一条四通八达的主动脉把业务库的binlog日志、用户行为埋点、服务器日志、甚至第三方接口回调统一收拢、清洗、分发到不同的目标端。这篇文章我就想把自己在实时数据同步方案上的选型思路、架构设计、参数调优和踩坑经历掰开揉碎讲一讲。无论你是刚接手实时同步任务还是已经在维护Kafka集群应该都能找到点有用的东西。1. 实时数据同步的整体思路与选型逻辑1.1 同步场景的分类为什么选Kafka作为中枢大数据场景下数据同步这个词其实涵盖了好几种不同形态的需求。最常见的是关系型数据库之间的同步比如把MySQL的业务数据同步到ClickHouse、Doris或者数据仓库Hive还有一种是日志类数据的同步比如App的埋点日志、Nginx访问日志需要通过日志采集器送到Kafka再落到HDFS或者对象存储做分析另外还有缓存、搜索引擎这类数据同步比如更新Redis缓存、构建Elasticsearch索引的时候需要把数据变更实时推送给下游。这几种场景有个共同点数据产生方和数据消费方的处理能力是不同步的。业务库的写入高峰可能瞬间冲到每秒几千上万条但下游的报表系统、推荐系统、搜索服务不一定能承受同样的压力而且每个下游关心的数据形态也各不相同。如果让业务系统逐条去调用下游接口或者让下游定时去业务库轮询要么把源系统拖垮要么数据的实时性上不去。Kafka在这里面扮演的就是一个削峰填谷的角色。所有数据先进入Kafka数据在Topic里按分区顺序存储、按消费者组并行消费生产者和消费者之间完全解耦。这样业务系统只需要把变更数据写进Kafka下游需求方各自按照自己的消费速率去拉取。等到实时数仓、数据湖、微服务这些概念流行起来之后Kafka几乎成了标准的数据中枢选型。1.2 主流同步方案对比直连、轮询与日志抽取真正落到方案选型的时候常见的有三大类应用双写、定时轮询、基于日志的变更捕获。应用双写就是业务代码在写数据库的同时再自己发一条消息到Kafka或者调用下游接口。这种方式实现简单但侵入性太强业务代码里到处埋点而且很容易出现数据库成功、消息发送失败这种事务一致性问题。定时轮询则是每隔一段时间扫描业务表的新增或修改记录实现成本低但实时性差而且每次扫描都对源库造成压力如果表上缺少合适的索引全表扫描能把数据库拖到告警。现在主流方案几乎都是基于日志的变更捕获。源数据库的binlog或者WAL日志记录了每一次数据变更的原始痕迹通过Canal、Debezium这类工具读取日志解析成统一的JSON或AVRO格式再写入Kafka下游拿到数据后自行落库或加工。这种做法对源库几乎无侵入也不要求业务表有额外的更新时间字段实时性可以做到秒级以内。我当时给某电商项目做用户订单数据同步正是切换到这个方案之后才真正告别了每天凌晨批处理带来的高延迟。用Kafka做同步中转其实还顺手解决了一个老问题多消费者之间的数据一致性。多个下游订阅同一个Topic彼此之间互不影响各自维护自己的消费位点谁慢了谁追进度不会因为一个下游挂了就阻塞其他人。这是直连或者单点推送做不到的。2. 核心架构从源端到目标端的链路设计2.1 一个典型的实时同步链路长什么样一套成熟的同步架构通常分成四层源端采集层、消息中间件层、目标端写入层、监控与元数据管理层。源端采集层负责把不同来源的数据统一成标准格式。MySQL走Canal或Debezium解析binlogPostgreSQL走wal2json或Debezium日志文件走Filebeat或FluentdApp埋点通过HTTP接口上报到采集网关。所有数据到了Kafka之后以Topic为粒度进行分类存储。这里要注意一下很多人在第一步就掉坑了Topic的命名和数据的格式没有提前规划。比如订单数据可能来自订单服务、支付服务、退款服务如果每个服务都往独立的Topic里发原始对象下游想组装一条完整的订单生命周期数据就要同时消费好几个Topic再做JOIN。我们后来把Topic按业务域统一规划成了类似order_base_info、order_pay_flow这样的命名方式格式也统一成包含业务主键、操作类型INSERT/UPDATE/DELETE、变更前数据、变更后数据、采集时间戳的通用信封结构下游用起来省了很多事。目标端写入层就是各类消费者程序从Kafka拉取数据执行ETL抽取、转换、加载写入目标存储。比如订单数据写入Doris做实时OLAP用户标签写入Redis供在线推荐服务读取行为日志写入Hive做离线分析。监控与元数据管理层则是很多团队容易忽略但实际上救过大命的一层。Kafka自身的指标、消费者的Lag、Sink任务的写入成功率、格式解析的失败量全部要纳入监控。没有这层消息积压到几百万条时你才发现问题下游已经延迟了半小时。2.2 Topic与分区设计的关键考量分区数怎么定是Kafka同步方案里最需要反复斟酌的事。先讲概念。Kafka的一个Topic会被物理切分成多个分区每个分区内部保证数据的有序性。分区是并行度的基本单位消费者组里同一时刻每个分区只能被一个消费者线程处理所以分区数决定了消费的最大并行度Producer写入时按照消息的Key哈希到不同分区同一个Key的数据永远进同一个分区这样保证了局部有序。分区数太少消费者并发上不去吞吐受限分区数太多又会带来一堆副作用比如文件句柄数暴涨、集群元数据变大、分区rebalance时间变长。我见过的生产事故里因为分区数拍脑袋设了40个、64个后来又因为业务扩张到上百个消费者线程rebalance频繁到每分钟来一次消费基本处于停滞状态。我自己的经验是分区数先按目标端的预估写入并发来定。比如下游是Doris集群期望20个并发写入任务那Topic分区数大概在20~40之间比较合理。不要听网上说的分区数越多吞吐越大先把单消费者的数据处理能力跑出来再倒推分区数。我用3000万条订单数据实测过单消费者每秒处理8000~12000条目标端写入MySQL瓶颈在5000条每秒左右那20个分区配合20个消费者就足够支撑每秒10万条级别的同步了。之后再去动态加分区很容易打乱之前的分布。另外分区数还牵扯到Key分布。如果业务上需要同一订单的所有变更消息按顺序被下游处理那Producer发送时就必须指定订单ID作为Key这样同一个订单的变更一定进入同一个分区。如果Key不指定所有消息轮询发往各个分区顺序就是乱的下游做数据更新的时候极容易出现旧数据覆盖新数据的惨剧。2.3 目标端写库的几种消费模式Kafka消费端写目标库业界常见的模式有三种逐条同步写、批量攒批写、基于事务的两阶段提交。逐条同步写最简单拉一条消息就执行一次INSERT或者UPDATE逻辑清晰适合数据量小、对实时性要求极高的场景。但问题在于和数据库的每一次交互都有网络开销吞吐上不去而且一旦数据库瞬时抖动客户端超时重试就会造成大量重复写。批量攒批写是目前主流的高吞吐方式。消费者在内存里维护一个数据缓冲积累到一定条数比如500条或者一定时间窗口比如2秒一次性通过JDBC的rewriteBatchedStatements批量写入目标库。这种方式能把写库吞吐提升好几倍。需要注意的事项是批大小别设置得太夸张我见过有人把批量设到5000条结果目标库的undo log直接暴涨最后整个数据库卡死。一般来说1000条左右是一个比较安全的量级根据实际压测微调。基于事务的两阶段提交则是在消费者处理完一批消息后把这笔数据处理完成的确认信息和目标库的写入操作放到同一个事务里。比如目标库是MySQL可以在事务里执行业务SQL同时更新Kafka消费位点表。这种模式下数据一致性的级别最高但吞吐会受限于事务大小和数据库负载一般用于对账、金融交易这类绝对不能丢数也不能重复入账的场景。3. 实操环节各环节关键参数与执行明细3.1 部署与基础参数配置从单机到集群先聊部署。如果你只是做学习和Demo单机部署一个Kafka用KRaft模式代替ZooKeeper一条命令就能起来。但生产环境的实时同步链路至少要三台机器起步奇数节点是为了故障容错时选举Leader方便。机器配置建议内存越大越好Kafka依赖操作系统页缓存JVM堆其实不需要给特别大关键是页缓存要留给活跃的Segment文件。我们当时的集群是12台机器每台32GB内存JVM堆设了8GB剩下24GB全部留给页缓存线上跑了几万分区也没有出现内存瓶颈。生产环境的服务端参数有几个必须重点关注。log.retention.hours控制消息保留时间同步场景建议至少保存24小时以上这样下游任务出问题或者新消费者要从头消费时还有补救余地我们设置的是72小时。log.segment.bytes默认值是1GB如果你的业务单条消息体积偏大比如日志类数据动不动上百KB建议把segment调小到256MB否则单条消息超过segment上限会导致整个分区无法写入。min.insync.replicas必须设成2配合发送端acksall才能保证Leader挂了之后数据不丢这点在同步方案里属于底线级别的配置。3.2 生产者端如何保证数据不丢不重生产端同步数据到Kafka最大的关注点就是三条不丢、不重、不乱序。不丢的核心是发送确认机制。Producer的acks参数有三个取值0表示发送出去就不管结果1表示Leader写成功就算成功all表示ISR里所有副本都写成功才算成功。同步场景建议必须用acksall。同时配合min.insync.replicas2这样即使一个副本所在机器宕机数据仍然有另一个副本存在不会丢。不重就稍微难一点。Kafka本身通过Producer的幂等机制enable.idempotencetrue可以保证同一个Producer会话内每条消息虽然可能因网络重试而被发送多次但Broker端只会持久化一次。注意这个幂等只针对单个Producer实例如果你的应用是多个进程并发写同一个Topic那还是可能产生重复消息。所以在发送端写一个全局唯一的MessageID下游再按照MessageID做去重才是治本的方法。不乱序的核心是单分区有序。幂等Producer开启之后如果没有指定KeyKafka默认按粘性分区策略往多个分区轮询写那消息整体顺序就不能保证。我们做订单状态同步的时候在Producer里强制按业务主键取哈希保证同一个订单的消息永远进同一个分区。这里还有一个细节如果某条消息发送失败重试它会尝试发到同一个分区但会排在在它之后已成功消息的后面这就是max.in.flight.requests.per.connection参数的影响。为了保证严格顺序要把这个参数设置成1也就是说同一个连接上只能有一个未确认的请求在途。代价是吞吐下降好在多数同步场景单分区吞吐本来就够用。3.3 消费者端并发、事务与幂等控制消费者端是整个同步链路里坑最多的地方因为消费逻辑出了错影响的是下游所有系统。第一个要聊的是消费者组的并发模型。同一个消费者组里所有消费者实例共同消费一个Topic下的所有分区组内的消费者数量不能超过分区数否则多余的消费者只是闲置。我之前犯过一个低级的错误Topic 32个分区消费者组里却起了40个线程的Sink任务结果8个线程空转数据全部积压在那32个分区的消费线程里。后来改成消费者数等于分区数每个消费者内部再做多线程批处理问题才解决。第二个重点是消费者参数。max.poll.records决定单次poll返回多少条消息默认值是500。如果你的单条消息体量大或者下游处理逻辑复杂这个值需要调小否则一次poll的数据量太大处理时间超过了max.poll.interval.ms默认300秒消费者就会被判定为挂掉被踢出组触发Rebalance。这个参数我强烈建议按你处理一批数据的平均耗时去设置留足缓冲。我们生产上设置的是max.poll.interval.ms600000因为有个下游Sink做数据清洗时比较重曾经出现过处理耗时超过300秒触发rebalance、rebalance之后又重复消费的循环。第三个是消费位点提交策略。自动提交enable.auto.commit如果开着默认每5秒提交一次offset这意味着你处理完数据但还没提交时挂了重启之后就会从上次提交的位置重新消费造成重复如果处理过程中提交了但数据实际还没写入目标库那就会丢数据。所以同步场景下我建议关闭自动提交手动在数据处理成功之后调用commitSync或者commitAsync。配合之前说的下一位点表加目标库事务能做到真正的不丢不重。幂等控制方面最土的方案就是在目标表上建唯一键写入时用INSERT ON DUPLICATE KEY UPDATE或者MERGE语句。但这个方案对UPDATE和DELETE场景不友好因为数据库的binlog同步到Kafka时一条DELETE消息到了下游你如果用INSERT的方式去执行目标库里这条数据根本不存在导致更新行数为0。所以更稳妥的做法是让目标表带上一个变更版本号每次写入时比较版本号大小只有更大版本号的数据才允许覆盖这种方法我们在ES同步里用得非常顺手。4. 工具选型Connect、Debezium、Canal到底怎么选4.1 各工具的真实定位与适用边界聊完手写的消费端再聊一聊现成的工具。同步链路里数据从源库到Kafka这一段现在最常用的工具就是Canal、Debezium、Flink CDC和Kafka Connect。Canal是阿里巴巴开源的产品主打MySQL binlog解析。它对MySQL的支持特别好部署相对简单配置一个canal.properties就能跑起来。但问题在于它只能解析MySQL不支持PostgreSQL、Oracle等数据源而且社区活跃度和新版本迭代这几年明显减缓。如果你的源库比较单一就是MySQL用Canal是可以的但你要有心理准备自己维护一些解析逻辑层面的小坑。Debezium是目前最活跃的CDC框架它本身是作为Kafka Connect的Source Connector运行的支持MySQL、PostgreSQL、MongoDB、SQL Server等多种数据库二进制日志解析和Schema变更的兼容性做得非常到位。它输出的消息格式里包含了op字段c表示创建、u表示更新、d表示删除以及before和after字段下游拿到之后很容易分辨操作类型。Debezium还有一个厉害的地方是支持Schema演进源表的字段新增了、类型变更了它会自动感知并更新输出格式Downstream不会因为字段对不上就解析失败。我个人建议如果数据库种类杂或者未来可能引入PG、MongoDB就直接上Debezium。Kafka Connect则是Apache Kafka自带的集成框架负责把外部系统数据导入KafkaSource或者把Kafka数据导出到外部系统Sink。用它来做同步场景的优势是原生整合Kafka生命周期管理和部署方式都很统一但它的生态里真正好用的Sink连接器质量参差不齐很多商业连接器是闭源的遇到问题排查困难。Confluent的JDBC Sink Connector确实很好用但多数是订阅制如果预算有限用Kafka Connect跑自研的逻辑会比较吃紧。Flink CDC就更偏流式计算了它是基于Flink框架的流式数据处理工具从数据库读取binlog后可以直接在Flink里做JOIN、聚合、窗口计算再把结果写出去。如果你的同步链路里还包含实时计算加工用它省掉一层Kafka中转是划算的但如果你只是想单纯把MySQL数据原样同步到数仓直接用Flink CDC有点杀鸡用牛刀的感觉而且它对状态后端、checkpoint机制的要求让运维变得更复杂。4.2 我的选型建议与实际落地组合我自己经历过一次选型翻车。早期用Canal把MySQL的binlog同步到Kafka规模小的时候跑得特别稳。后来公司引入了PostgreSQL作为新的业务库Canal不支持PG临时加了日志采集工具绕了一圈搞得整个链路混乱得不行最后下决心统一迁移到了Debezium。如果是新起一个同步项目我更推荐Debezium Kafka Connect这种组合。它不需要业务应用嵌SDK部署几个Connector进程就行通过REST API就能动态管理连接任务扩缩容也很方便。配合Kafka Connect的Single Message Transform功能还能直接在采集阶段做轻量级的字段过滤、类型转换和消息路由避免下游重复处理这些脏活。目标端写库这一段我现在的经验是能用现成Sink就用现成Sink但不能指望它处理所有脏数据。比如Doris官方有提供StreamLoad相关的Connector性能很好直接消费Kafka写进Doris就行。但ES的Sink即便用官方插件遇到字段类型冲突也是会报错的这时候还不如自己写一个消费程序控制字段映射的细节出了问题也好查日志。5. 常见问题与排查技巧实录5.1 消息积压严重消费Lag爆炸式上升实时同步场景最经常遇到的告警就是消费者Lag飙升。所谓的Lag就是消费者消费到的offset和生产端最新写入的offset之间的差值差值越大说明数据积压越严重。第一步肯定是要定位是哪个环节慢了。我会先用脚本查看每个消费者组每个分区的Lag分布如果某个分区的Lag特别高而其他分区正常那大概率是数据倾斜也就是说这个分区的消息体量本身特别大或者某条消息处理极慢如果所有分区Lag都高那就是消费者整体吞吐不行或者下游写入端出现瓶颈。前者要检查Key的分布逻辑或者单独扩容那个分区的消费并发后者要重点排查目标库比如MySQL的慢查询、锁等待ES集群的bulk队列被打满Doris的BE节点CPU跑满。还有一次积压是因为我们生产端的压缩设置引起的。Kafka的Producer开了snappy压缩理论上能降低网络带宽占用但如果下游目标库对单个字段的长度有限制比如一次批量写因为某条数据超长而反复失败重试整个消费循环就卡住了。排查了半天才意识到是压缩后字段内容过大。后来我们在Producer端增加了max.request.size的上限控制同时在下游做了字段长度截断问题才解决。说到排查积压还有一个很容易被忽略的地方消费端执行了长时间GC。JVM的Full GC暂停几十秒消费者相当于停摆Lag自然猛涨。用G1垃圾回收器切实调大MaxGCPauseMillis之后这类问题就很少出现了。5.2 数据重复消费与下游重复写入的应对重复消费是Kafka同步躲不开的话题。只要消费者在提交offset之前崩溃或者Rebalance导致分区重新分配都会造成一部分消息被重复消费。网络层面的重试导致消息重复投递也会增加重复概率。我处理重复数据的招数是从两条线同时下手Kafka端尽量保证不重复下游端尽量做到可重入。Kafka端的措施主要有关闭自动提交改成业务逻辑执行成功后手动提交重平衡监听器里做优雅处理在分区被回收前把当前批次数据处理完并提交开启幂等Producer防止同一消息在Broker端物理重复存储。下游端的措施主要是唯一键和版本号。比如目标Doris表里设置一个unik_key写入时通过StreamLoad的upsert语义去重目标MySQL表建唯一索引SQL里加上ON DUPLICATE KEY UPDATE只更新指定列。ES场景下我们会在业务主键前面拼上Kafka的分区号和offset这样即使上游重发了同一条逻辑消息ES文档的_id也是一样的覆盖写进去的仍然是同一条记录。5.3 数据乱序为什么下游拿到的是旧数据乱序这个问题我曾经在一次订单状态同步的事故里栽过跟头。业务方反馈用户下单之后查看订单详情偶尔会看到几分钟之前的旧状态后来又变成新状态。排查下来才发现问题出在Kafka的同一分区里出现了乱序。原因是我们的Producer应用是多个实例部署的虽然消息都指定了订单ID作为Key理论上同一订单的变更消息会落在同一个分区但两个Producer实例各自开启幂等在Broker端收到消息的先后顺序并不完全等同于业务发生的顺序。比如订单从已创建变到已支付第一条消息因为网络问题发送失败了Producer自动重试这期间第二条已支付消息已经成功写入了分区等第一条重试成功之后分区的顺序就反了下游先收到已支付再收到已创建然后旧状态把新状态覆盖了。解决办法有两个方向。方向一是在Producer内部加一个本地队列按业务Key做排序后逐个发送牺牲部分吞吐适合对顺序要求极其严格的场景。方向二是在数据信封里带上业务时间戳下游处理时比较版本号只有当新的消息版本大于旧消息时才覆盖写入。我强烈建议用方向二因为即使你在生产端做到顺序完美下游如果有并行任务的场景顺序仍然可能被打破所以兜底逻辑必须有。5.4 Schema变更引发的消费中断这是CDC同步特有的麻烦。源库的某个表增加了字段或者把字段从varchar改成了bigint如果下游的写入SQL还是老的结构解析出来的消息和写入目标之间就会报字段不匹配。Debezium在发现源库Schema变化之后默认行为会结合配置决定是继续还是暂停任务。我们在配置里设了schema.history.internal来保存历史Schema但当目标端是宽表结构或者源端的字段改名之后下游的映射逻辑还是需要手动改。这里有个经验是同步任务里要加一个运行时的字段映射规则配置不要写死在代码里。把源字段名、目标字段名、类型转换函数作为外部化配置每次源库Schema变更只要更新配置任务就能热加载不需要重启消费进程。我们后来把这类配置放到配置中心再用一个告警规则监听Debezium的Schema变更事件基本能做到分钟级响应。5.5 一套接地气的监控指标体系最后说一下监控这块。Kafka自带了一些JMX指标但默认不会暴露成可操作的告警项。我建议至少盯这几个指标消费Lag、生产/消费请求的吞吐TPS、分区Leader的分布是否均匀、Broker的磁盘使用率、消费者的poll处理耗时、目标端写入失败量。消费Lag是核心指标不必多说吞吐TPS可以用来判断系统当前有没有接近瓶颈分区Leader分布不均匀说明某个Broker承担了太多读写压力要进行分区重平衡。磁盘使用率是最大的隐患Kafka如果不开启日志自动清理磁盘满了很可能会挂掉整个Broker。我们线上就是配置了磁盘使用率达到70%就告警80%触发自动清理掉过期日志这才把磁盘写满的事故遏制在萌芽里。写到这里最想跟各位分享的还是自己的一个体会做实时数据同步不要总想着追求完美的一致性方案而要根据业务场景去平衡吞吐、延迟、可靠性这三者的关系。交易类数据要严格、宁可慢也不能错日志分析类数据追求吞吐和成本搜索索引同步则更看重端到端延迟。我在实际项目中反复调整过很多次参数也换过好几轮工具但始终没变的思路是先把同步链路的可观测性做好任何问题发生都能快速定位到源端、中间件还是目标端再谈优化。毕竟一把好用的监控和一套清晰的排查流程往往比任何一种完美架构更能救你于水火。
阅读完成 · 觉得有帮助？
