4、实时数据管道搭建实战：从数据源采集到Kafka、使用Flink进行简单的ETL处理、将结果写入Redis/MySQL
这一章，我们直接动手。
前面聊了那么多概念和架构，现在终于要落地了。实时数据管道，说白了就是把数据从源头捞出来，经过清洗和转换，最后送到目标存储里。整个过程要快，要稳，还不能丢数据。
我刚开始做实时管道的时候，踩过不少坑。有一次线上数据延迟了十几分钟，排查了半天才发现是Kafka的消费者配置有问题。嗯，从那以后我对每个环节的参数都格外小心。
本章核心目标：搭建一条完整的实时数据管道，涵盖数据采集、消息队列、流处理和结果存储四个环节。
4.1 整体架构概览
先看一张图，把整个流程串起来。你想想看，数据从产生到落地，到底经历了什么？
整个管道分四步走：
- 数据源采集：从业务数据库或日志文件中捕获变更数据
- 消息队列：用Kafka做缓冲和解耦，削峰填谷
- 流处理：Flink负责清洗、转换和聚合
- 结果存储：写入Redis做缓存，或写入MySQL做持久化
4.2 数据源采集：从MySQL Binlog说起
实时管道的第一步，是拿到数据。最常见的场景是监听MySQL的变更。
我个人习惯用Canal来做这件事。Canal会伪装成MySQL的从库，实时接收Binlog的变更事件。你想想看，每次有数据插入、更新或删除，Canal都能第一时间感知到。
小提示：生产环境中记得开启MySQL的Binlog，并且设置格式为ROW。否则Canal拿不到完整的变更数据。
配置示例：
# canal.properties
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal123
canal.instance.connectionCharset=UTF-8
canal.instance.filter.regex=test\\..*
# 监听test库下的所有表如果是日志类的数据源，比如Nginx访问日志，我会用Filebeat。它轻量、资源占用低，适合部署在每台服务器上。
4.3 Kafka：管道的核心枢纽
Kafka在这里扮演的角色，说白了就是一个超级缓冲区。采集层把数据扔进来，Flink再从里面拉走。两边互不干扰，各自按自己的节奏跑。
我记得有一次业务流量突然暴涨，Flink处理不过来。但因为Kafka扛住了，数据一点没丢。等Flink恢复后，自动从断点继续消费。这就是消息队列的价值。
Kafka的关键配置：
| 参数 | 推荐值 | 说明 |
|---|---|---|
| num.partitions | 3~6 | 分区数，影响并行度 |
| replication.factor | 2~3 | 副本数，保证高可用 |
| retention.ms | 604800000 | 数据保留7天 |
| acks | all | 确保数据不丢失 |
⚠️ 注意：分区数不是越多越好。分区太多会导致文件句柄暴增，反而影响性能。我一般建议按业务流量预估，留20%的余量。
4.4 Flink ETL：清洗与转换
数据到了Flink，就该干活了。ETL说白了就三件事：过滤掉脏数据、转换字段格式、做简单的聚合。
我曾经接手过一个项目，上游数据里混入了大量空值和乱码。如果不做过滤，下游的Redis和MySQL会直接报错。所以，第一件事永远是数据清洗。
一个典型的Flink ETL作业：
// 1. 从Kafka读取数据
DataStream<String> sourceStream = env
.addSource(new FlinkKafkaConsumer<>("input-topic",
new SimpleStringSchema(), kafkaProps));
// 2. 数据清洗：过滤空值和格式异常
DataStream<UserEvent> cleanStream = sourceStream
.filter(line -> line != null && line.contains("user_id"))
.map(line -> parseToUserEvent(line));
// 3. 数据转换：字段补齐、格式统一
DataStream<UserEvent> transformedStream = cleanStream
.map(event -> {
event.setTimestamp(System.currentTimeMillis());
event.setStatus("processed");
return event;
});
// 4. 简单聚合：按用户ID统计事件次数
DataStream<UserCount> aggregatedStream = transformedStream
.keyBy(event -> event.getUserId())
.timeWindow(Time.seconds(10))
.apply(new CountWindowFunction());核心原则：Flink的ETL尽量保持无状态。如果必须做状态管理，记得设置合理的TTL，避免状态无限膨胀。
4.5 结果写入：Redis和MySQL
处理完的数据，最终要落到存储里。这里分两种情况：
- Redis：适合做实时缓存、排行榜、计数器。写入速度快，但数据量不宜过大。
- MySQL：适合做持久化存储，方便后续查询和分析。写入速度相对慢，需要控制写入频率。
Flink写入Redis的示例：
// 使用Flink Redis Connector
DataStream<Tuple2<String, String>> redisStream = transformedStream
.map(event -> new Tuple2<>(event.getUserId(), event.getEventType()));
redisStream.addSink(new RedisSink<>(redisConfig, new RedisCommandDescription(RedisCommand.SET)));Flink写入MySQL的示例：
// 使用JDBC Sink，批量写入
aggregatedStream.addSink(JdbcSink.sink(
"INSERT INTO user_counts(user_id, count, window_end) VALUES (?, ?, ?)",
(statement, count) -> {
statement.setString(1, count.getUserId());
statement.setLong(2, count.getCount());
statement.setLong(3, count.getWindowEnd());
},
JdbcExecutionOptions.builder()
.withBatchSize(1000)
.withBatchIntervalMs(5000)
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/test")
.withDriverName("com.mysql.cj.jdbc.Driver")
.withUsername("root")
.withPassword("root123")
.build()
));经验之谈：写入MySQL时，一定要用批量写入，不要一条一条插。我见过有人逐条插入，结果Flink的吞吐量直接掉到几百条每秒。改成批量后，轻松上万。
4.6 避坑指南
做实时管道，坑不少。我挑几个最常见的说说：
- 数据重复：Flink的Exactly-Once语义需要配合Kafka的事务和检查点。如果配置不当，可能会收到重复数据。我建议在目标存储里做幂等处理。
- 背压问题：当Flink处理速度跟不上Kafka的写入速度时，会出现背压。这时候要么加资源，要么优化算子逻辑。
- 连接超时：Redis和MySQL的连接池要设置合理。我曾经因为连接池太小，导致Flink频繁重连，性能骤降。
⚠️ 重要提醒：生产环境一定要开启Flink的Checkpoint。否则一旦作业失败，数据会从Kafka的最早偏移量重新消费，造成大量重复甚至数据错乱。
好了，这一章的内容就到这里。管道搭起来了，数据也跑通了。接下来，你可以根据自己的业务场景，调整ETL逻辑和存储策略。记住，实时管道没有银弹，适合自己业务的才是最好的。