Spark Streaming+Canal+Kafka：MySQL增量数据实时监测分析实战

做实时数据监测项目时，最头疼的问题往往不是怎么写统计SQL，而是怎么把MySQL里的每一次变更稳定地搬到下游。当时某系统的业务库有几百张表，凌晨批处理报表越跑越重，业务方又要求最快速度看到增量数据，于是只能围绕Canal、Kafka、Spark Streaming这套组合搭一条MySQL增量数据实时监测链路。这条链路并不只是简单采集binlog，而是把binlog变成流式事件，再通过Spark做实时分析，为数据同步、监控大屏、异常预警提供指标。

如果你打算做“基于Spark Streaming + Canal + Kafka对MySQL增量数据实时进行监测分析”，或者只是想搞清楚这几个组件放到一起到底怎么玩，这篇文章会给你一套能落地的思路。我尽量按实际踩坑的顺序讲，先聊选型，再说配置，最后给出一段可以直接参考的Spark代码，末尾整理了几个我印象很深的问题。

1. 增量数据链路整体规划与选型思路

1.1 为什么把Canal放在采集层

MySQL的增量数据来源只有一条，binlog。所有针对MySQL数据变更的实时采集，本质上都是想办法伪装成MySQL的从库，去订阅binlog事件。Canal做的就是这件事，它把自己模拟成只读从库，解析完binlog后通过客户端或消息队列往外发。

市面上类似工具也不少，比如Maxwell、Debezium。我选择Canal的核心原因是它和MySQL贴得近，对binlog的解析相当成熟，部署结构也简单。Canal sserver可以独立跑，也可以配集群，配合ZooKeeper做高可用。如果只是想监控某个业务库的表变更，Canal是最快出效果的方案。

需要提醒的是，Canal不是一个通用ETL平台，它只负责“把binlog变成可消费的数据”，不负责转换。真正的清洗、聚合和指标计算还是要交给下游流处理引擎。一个最开始容易犯的错，是希望Canal把所有表变更都在采集端拼成标准格式，那会让配置变得臃肿，还不如让它保持通用，后续用Spark做事件解析。

1.2 Kafka在链路中扮演什么角色

从Canal直接拿到MySQL增量数据，听起来够快，但生产环境不会让业务系统直接订Canal客户端，原因很简单：Canal服务一旦重启或者网络抖动，数据会积压，下游处理节点也会跟着抖动。

Kafka在这里的价值是削峰和解耦。Canal往Kafka写入，Spark从Kafka自由消费，二者互不阻塞。binlog高峰期可能一分钟上万条事件，Kafka能扛住；Spark服务重启时，数据还会按Topic保留一段时间，不会丢。链路变成MySQL -> Canal -> Kafka -> Spark之后，每一层都能独立扩容。

Kafka Topic的规划也很重要。通常我会按业务域拆Topic，不要所有库表混成一个超大Topic。某项目刚开始把所有binlog丢进同一个Topic，Spark那边要反复过滤无效事件，白白浪费很多CPU。合理一点的做法是：核心业务表单独Topic，日志类表放一起，低频变更表放一起。

1.3 Spark Streaming与Structured Streaming如何取舍

很多方案文档会把这一整段称为Spark Streaming，但实际新项目里我更推荐使用Structured Streaming。原因很直接：Structured Streaming基于DataFrame/DataSet API，能像写批处理一样写流处理，处理窗口、事件时间、状态管理都有更现成的功能，而且和Structured Streaming本身的sink机制配合更好。

如果你碰到一个老项目，里面还在用DStream API，那只说明项目创建时间比较早。DStream基于RDD，写起来琐碎，操作每个批次数据都要额外转换。真遇到状态管理或者延迟比较高的窗口计算，会比较痛苦。

我并不是说DStream彻底不能用。如果你有一些奇怪的算子需求，或者团队对RDD API非常熟，用DStream也可以。但对于本文这类的增量监测分析场景，Structured Streaming明显更顺手。所以我下面的实现部分，会直接用Structured Streaming写法，同时保留Spark Streaming作为搜索关键词的表达习惯。

2. 链路搭建前的关键参数配置

2.1 MySQL侧必须改的核心参数

要让Canal拿到完整增量数据，MySQL必须提前打开binlog，并采用Row格式。下面这组参数建议写进MySQL配置文件，然后重启MySQL实例。

有一点容易忽略，Canal连接MySQL的账号最好只开通必要权限，不要直接用root。新建账号时给SELECT、REPLICATION SLAVE、REPLICATION CLIENT三个权限就够了。MySQL 8.0默认密码插件是caching_sha2_password，Canal不一定支持，通常需要指定mysql_native_password方式创建账号。

2.2 Canal服务端配置与Kafka对接

Canal启动后可以配置成纯TCP模式，但那样就需要自己写客户端去连。针对Kafka链路，更推荐直接把Canal接入Kafka。Canal 1.1.4之后的版本，配置文件里有个核心开关canal.serverMode，设置成kafka后，解析出的数据会投递到Kafka Topic。

canal.properties里常见的配置项如下：

canal.id=1 canal.ip=0.0.0.0 canal.port=11111 canal.serverMode=kafka canal.mq.servers=192.168.10.10:9092,192.168.10.11:9092 canal.mq.retries=0 canal.mq.batchSize=16384 canal.mq.maxSize=1048576

instance.properties里面指定的是MySQL实例信息，以及要监听哪些表。比如我们只想监听app库下面user_order这张表：

canal.instance.master.address=192.168.10.20:3306 canal.instance.dbUsername=canal_user canal.instance.dbPassword=canal_pass canal.instance.filter.regex=app\\..* canal.mq.topic=binlog_event

这里canal.mq.topic如果设置成动态topic，Canal会把库名表名拼进去，比如默认就是application_表名。如果只需要一个统一Topic，可以直接写死一个名称。我建议第一次搭建用固定Topic，先把链路跑通再说。

2.3 Kafka主题规划与消息体格式

Kafka Topic分区的数量直接影响Spark消费并行度。Canal写入Kafka时，也会按分区规则散列。为了方便乱序，最简单的办法是让Canal按表名哈希分区。这样同一张表的数据大概率落进同一个分区，Kafka分区内部是有序的，Spark消费时顺序更有保证。

创建Topic时，我一般会把分区数定为下游Spark并发数的整数倍。比如Spark分配4个executor，每个executor开2个并发线程，那么分区数可以设为8。不要急着设成100个分区，分区越多，Kafka集群内部开销越大。

Canal输出给Kafka的消息格式有两种，默认自带protobuf格式比较省流量，但调试麻烦。如果接Spark做分析，建议打开Canal的flatMessage开关，让消息以JSON字符串形式写入Kafka，样例大概长这样：

{ "data": [{"id": 1001, "name": "张三", "order_amount": 99.00}], "old": null, "type": "INSERT", "database": "app", "table": "user_order", "es": 1698888888000, "ts": 1698888889000 }

看到这个JSON结构就知道，Spark要做的工作很清晰：解析JSON、过滤事件类型、提取data字段、按库表和事件类型做聚合。

3. Spark Streaming消费增量数据的具体实现

3.1 从Kafka读取增量流的入口配置

假设Kafka里的Topic名为binlog_event，下面是一段基于PySpark的Structured Streaming读取代码，很适合快速验证链路：

from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, explode spark = SparkSession.builder \ .appName("mysql_binlog_monitor") \ .master("yarn") \ .getOrCreate() df_raw = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "192.168.10.10:9092") \ .option("subscribe", "binlog_event") \ .option("startingOffsets", "latest") \ .option("maxOffsetsPerTrigger", 20000) \ .option("failOnDataLoss", "false") \ .option("enable.auto.commit", "false") \ .load()

这里面对不同的配置要解释一下。startingOffsets设置为latest，适合任务启动后只关心新产生的增量数据；但如果想从历史binlog追补数据，就要改成earliest或者指定JSON格式的offset。maxOffsetsPerTrigger用来限制每个触发周期最多消费多少条数据，这个参数在生产环境非常重要，它能防止启动瞬间Kafka积压数据把Spark压垮。

failOnDataLoss我平时习惯先设为false，因为测试阶段可能手动清掉过offset，设为true会直接报错停止任务；真正上线时再评估是否改回true，因为它能帮助发现数据未消费就被清理的情况。

3.2 解析Canal消息并转换成结构化数据

Kafka读出来的原始value是二进制，需要先转成字符串，再用from_json解析。建议先定义Canal消息的JSON Struct，不要每消费一条就写一个udf去解析，udf性能相对较差。

from pyspark.sql.types import StringType, StructType, StructField, LongType, ArrayType canal_schema = StructType([ StructField("data", ArrayType(StringType()), True), StructField("old", ArrayType(StringType()), True), StructField("type", StringType(), True), StructField("database", StringType(), True), StructField("table", StringType(), True), StructField("es", LongType(), True), StructField("ts", LongType(), True) ]) df_json = df_raw.selectExpr("CAST(value AS STRING) AS json_str") \ .select(from_json(col("json_str"), canal_schema).alias("msg")) \ .select( col("msg.database").alias("db_name"), col("msg.table").alias("table_name"), col("msg.type").alias("event_type"), col("msg.es").alias("event_time"), col("msg.ts").alias("process_time"), explode(col("msg.data")).alias("row_data") )

有一点要注意，Canal的data字段本来就是数组，因为一个事务里可能包含多行变更。用explode把数组展开后，每一行变更记录就会独立成为一条数据，后续统计才看得清变化量。

如果业务上需要区分老数据和新数据，再把old字段解析出来。old为null时表示新增，如果所有字段都有值，说明是更新操作。

3.3 增量实时监测分析的算子设计

拿到结构化事件流之后，常见的实时监测分析有几种：按事件类型统计变化量、按表和事件维度做窗口计数、识别高频变更表、统计操作延迟。

比如我想统计每分钟每张表的INSERT/UPDATE/DELETE数量，可以这样写：

from pyspark.sql.functions import window, count analysis_df = df_json.groupBy( window(col("process_time"), "1 minute"), col("db_name"), col("table_name"), col("event_type") ).agg(count("*").alias("event_count")) query = analysis_df.writeStream \ .outputMode("append") \ .format("console") \ .option("truncate", "false") \ .start()

如果要把结果实时写入MySQL，我会用foreachBatch。Structured Streaming允许一个批次作为一个DataFrame写到MySQL，可以复用已有的JDBC写入逻辑，也能避免用小批量sink把连接池打爆。

def write_to_mysql(batch_df, batch_id): batch_df.write \ .mode("append") \ .format("jdbc") \ .option("url", "jdbc:mysql://192.168.10.30:3306/analysis") \ .option("dbtable", "table_changes_minute") \ .option("user", "analysis_user") \ .option("password", "your_password") \ .save() query = analysis_df.writeStream \ .foreachBatch(write_to_mysql) \ .outputMode("append") \ .option("checkpointLocation", "hdfs:///spark_checkpoint/binlog_monitor") \ .start()

checkpointLocation必须指定。没有它，流任务重启后无法恢复offset。设了checkpoint之后，Spark会把offset和状态信息写到HDFS或本地目录，这是保证端到端不混乱的核心。

3.4 结果写入与报警触发

分析结果不只有写到数据库这一种落点。如果希望变更量超过阈值时立刻告警，可以在输出阶段加一个filter和foreach操作。比如某张核心表一分钟内DELETE超过五千条，就触发钉钉或短信报警。

alert_df = df_json.groupBy( window(col("process_time"), "1 minute"), col("table_name"), col("event_type") ).count().filter( (col("table_name") == "user_order") & (col("event_type") == "DELETE") & (col("count") > 5000) )

这里会涉及一个细节：Structured Streaming对foreachBatch之外的sink操作，如果还带有聚合，那么outputMode不能设为append，因为聚合结果会随数据更新而更新，需要用update模式或者automatic。实际项目里如果拿不准，就统一用foreachBatch，兼容性最强。

4. 生产环境常见问题与排查实录

4.1 Kafka收不到Canal推送的数据

链路搭好以后，最常遇到的就是MySQL有变更，Kafka消费端却什么都看不到。我的排查顺序基本固定：先看Canal实例状态，再看Canal日志，然后用Kafka自带Consumer验证Topic。

先确认Canal instance是否启动成功，在Canal安装目录下看到destination文件夹，检查instance.properties所在实例有没有报错。如果日志提示MySQL连接失败，多半是账号权限或地址不通。如果日志里出现canal.instance.mysql.slaveId冲突，说明server-id和现有MySQL节点重复，改一个不同的值就可以。

第二步是确认topic名是否对。Canal默认的topic自动拼接规则是instance名称或库名_表名，如果你在Spark中订阅的是固定的binlog_event，但Canal实际写入了app_user_order，那自然消费不到。统一用一条命令查看Topic列表：

kafka-topics.sh --bootstrap-server 192.168.10.10:9092 --list

然后再用Kafka客户端消费验证：

kafka-console-consumer.sh --bootstrap-server 192.168.10.10:9092 --topic binlog_event --from-beginning

能刷出JSON消息，说明问题不在Canal，而在Spark配置；刷不出消息，就要回过来检查Canal的配置文件。

4.2 Spark消费延迟持续增大的调优

Kafka里lag持续上涨，常见原因有三个：Spark处理太慢、并行度不够、下游sink阻塞。

先看并行度。如果Kafka Topic有8个分区，但Spark executor数量只有2个，每个executor哪怕开两个线程也只能一次读4个分区，另外4个分区闲置。把Spark并行度尽量对齐Topic分区数，会更高效。具体做法可以通过spark.sql.shuffle.partitions调整shuffle并行度，或者让executor数量乘以executor cores接近分区数。

再看maxOffsetsPerTrigger。如果你发现每个批次处理时间稳定，但Kafka消费来的量还是跟不上生产速度，说明上游确实大流量，这时可以适当调高maxOffsetsPerTrigger。但要注意，“调高”并不等于“越大越好”，因为Spark流处理背后是微批模式，一个批次数据量过大会导致GC压力上升。

最后检查下游写入。比如把分析结果写MySQL时，连接池太小或者表没有索引，一次写入几十万条记录就会慢出天际。合理做法是让下游表先建好索引，或者使用批量写入而不是逐行执行插入。

4.3 重启后数据重复或者丢失

Structured Streaming用checkpoint保存offset，正常重启不会丢数据。但有一种常见情况：在checkpoint目录被清理或者改动Topic分区数之后重启，Spark就会找不到有效的offset，数据要么重放，要么直接跳一段。

数据重复的另一个源头是幂等性设计。Spark精确一次只能保证处理状态存储层面的不重复，但它不保证写入MySQL不重复。你写入MySQL用的是append模式时，如果任务重启后重复执行同一个批次，MySQL中就会多出相同记录。

要解决这个问题，最简单的方法是在目标表加业务唯一键，写入时采用upsert语义，而不是append。MySQL没有原生的upsert，可以用INSERT ... ON DUPLICATE KEY UPDATE，或者把数据先落到临时表，再由批量任务合并。Financially，如果下游只是做分析报表，允许少量重复，那可以直接忽略。

4.4 几件让我印象很深的小事

第一件事是时区问题。Canal消息里的es和ts是毫秒时间戳，如果是GMT+8环境下，直接用Spark函数转成时间时，要显式指定时区，否则会差8小时。这类问题在实时报表里非常隐蔽，往往等到对账才发现。

第二件事是大事务导致超大消息。MySQL里一次性update一万行，binlog会产生大量事件，Canal会按照batchSize分批发送，但单个数据行本身如果包含很长的text字段，整条消息可能超过Kafka默认1MB。这时需要在Canal的canal.mq.maxSize和Kafka的max.message.bytes两端同时放宽，不然消费端会持续刷反序列化异常。

第三件事是不要在流处理里频繁物化全局状态。Structured Streaming做窗口聚合确实方便，但如果你要关联MySQL维度表，最好把维度表做成异步缓存，或者定期拉取到Redis，不要直接在每个批次里都执行一次JDBC查询。我见过一个项目把维度查询写在map算子里，流量一上来就接连超时，排查了很长时间才定位到JDBC连接等待。

根据我个人的经验，这套链路最好先从小流量、单表场景跑通，再快速推广到全库。很多问题都是全局铺开之后才暴露的，比如突然某个大事务把Kafka消息顶到几MB，某个表被大量update导致event_type倾斜，某个消费组在不该重置offset的时候被重置。遇到这些问题时，最重要的不是急着改代码，而是先观察Canal日志、Kafka lag和Spark流任务日志，定位到底卡在哪一层，再对症调整。

最后再分享一个实用小技巧：在用Canal采集MySQL增量时，不要只关注SQL语句，要关注binlog当前位点。上线前可以先跑一小时压测，确认MySQL主库binlog能正常增长、Canal能持续消费、Kafka分区消费速率稳定，再带着这套链路去接业务需求。按这个步骤一步步推进，增量数据实时监测分析这条路会平稳很多。