实时数仓的核心需求是数据低延迟同步，传统批量ETL往往存在小时级甚至天级延迟，无法支撑实时分析场景。Canal通过模拟MySQL从库的交互协议，能够实时订阅Binlog的增量变更，为实时数仓的ETL抽取提供可靠的数据源，再结合流式计算框架完成数据清洗、转换和加载，可实现端到端秒级延迟的数据同步。
Canal订阅Binlog的基本原理
Canal的核心工作流程分为三个步骤：首先模拟MySQL从库向主库发送dump协议请求，主库接收到请求后会推送Binlog日志给Canal；然后Canal解析Binlog日志，将二进制日志转换为结构化的变更事件，包含表名、操作类型、变更前后的数据等内容；最后Canal将解析后的事件推送到消息队列或者直接暴露给客户端订阅。
Canal支持多种Binlog解析模式，常用的有行模式，能够完整记录每一行数据的变更细节，适合实时数仓的精确数据同步需求。同时Canal提供了HA部署模式，通过ZooKeeper实现多节点协调，避免单点故障导致的数据订阅中断。
实时数仓ETL与流式计算的结合架构
完整的结合架构分为四层：数据源层为MySQL业务库，通过Binlog记录所有数据变更；数据采集层使用Canal订阅Binlog，将变更事件发送到Kafka消息队列；数据处理层使用Flink作为流式计算引擎，消费Kafka中的变更数据，完成数据清洗、维度关联、格式转换等ETL操作；数据存储层将处理后的数据写入实时数仓的对应分层，比如ODS层原始数据、DWD层明细数据。
Canal核心配置步骤
1. MySQL开启Binlog并配置权限
首先需要确保MySQL开启了Binlog，并且格式为ROW，修改MySQL配置文件my.cnf：
# 开启Binlog log-bin=mysql-bin # Binlog格式为ROW binlog_format=ROW # 服务ID，集群内唯一 server-id=1
然后创建Canal使用的MySQL账号，并授予复制权限：
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_password'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
2. Canal服务端配置
下载Canal部署包后，修改conf/example/instance.properties配置文件：
# MySQL主库地址 canal.instance.master.address=127.0.0.1:3306 # MySQL账号密码 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_password # 订阅的数据库和表，这里订阅test库下所有表 canal.instance.filter.regex=test\..* # 输出到Kafka的配置 canal.mq.topic=canal_binlog_topic canal.mq.bootstrap.servers=127.0.0.1:9092
启动Canal服务后，会自动订阅对应MySQL实例的Binlog，并将解析后的变更事件发送到Kafka的canal_binlog_topic主题中。
Flink流式计算处理Binlog数据
Flink消费Kafka中的Binlog变更事件，完成ETL处理后写入实时数仓，以下是核心代码示例：
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
public class CanalBinlogETL {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 配置Kafka数据源，消费Canal发送的Binlog主题
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
.setBootstrapServers("127.0.0.1:9092")
.setTopics("canal_binlog_topic")
.setGroupId("flink_canal_consumer")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
// 读取Kafka数据
DataStream<String> canalStream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
// 解析Binlog事件，完成ETL处理
DataStream<String> processedStream = canalStream.process(new ProcessFunction<String, String>() {
@Override
public void processElement(String value, Context ctx, Collector<String> out) {
try {
JSONObject canalEvent = JSON.parseObject(value);
// 获取操作类型，INSERT/UPDATE/DELETE
String opType = canalEvent.getString("type");
// 获取变更数据
JSONObject data = canalEvent.getJSONArray("data").getJSONObject(0);
// 获取表名
String tableName = canalEvent.getString("table");
// ETL处理：添加操作类型字段、处理时间字段
data.put("op_type", opType);
data.put("etl_time", System.currentTimeMillis());
// 过滤掉不需要的表数据，这里只处理user表
if ("user".equals(tableName)) {
out.collect(data.toJSONString());
}
} catch (Exception e) {
// 异常处理，可记录日志或发送到异常队列
}
}
});
// 处理后的数据可写入实时数仓，比如写入Kafka或者HBase、ClickHouse等存储
processedStream.print();
env.execute("Canal Binlog ETL Job");
}
}
注意事项与优化建议
- Binlog保留时间需要合理设置，避免Canal订阅延迟导致Binlog被清理，建议保留7天以上
- Canal的批量拉取大小和解析线程数需要根据MySQL的写入量调整，避免解析延迟
- Flink消费Kafka时需要设置合理的并行度，匹配Binlog的产生速率，避免数据积压
- 实时数仓的分层设计需要和Binlog订阅的粒度匹配，ODS层尽量保留原始变更数据，方便后续回溯
- 需要处理Binlog的DDL变更事件，避免表结构变更导致ETL任务失败，可在Flink中添加DDL事件处理逻辑
通过上述方案，可实现从MySQL Binlog订阅到实时数仓数据加载的全链路低延迟同步，满足实时报表、实时风控等场景的数据需求，相比传统批量ETL，数据延迟可从小时级降低到秒级，大幅提升实时数仓的时效性。