Flink MySQL CDC datastream

<think>我们正在处理用户关于Flink MySQL CDC DataStream API的使用和问题排查的请求。根据引用[4]，我们知道Flink MySQL CDC在启动时会获取全局读锁，然后读取binlog位置和schema，之后释放锁并开始从binlog读取变更。同时，它通过checkpoint机制保证exactly-once语义。 用户之前已经了解了Flink SQL CDC的配置，现在转向DataStream API的使用。因此，我们需要提供DataStream API的指南和常见问题排查。 步骤： 1. 环境准备：与SQL类似，需要MySQL开启binlog，并配置ROW模式。 2. 添加依赖：与SQL相同，需要添加flink-connector-mysql-cdc依赖。 3. 使用DataStream API创建源。 4. 处理数据并写入目标。 5. 问题排查。 注意：引用[1]提到Flink SQL CDC在1.11开始支持，而DataStream API也是基于相同的底层机制。 下面我们编写DataStream API的使用指南和问题排查。</think>### Flink MySQL CDC DataStream 使用指南与问题排查 以下基于 Flink 1.18+ 和 MySQL CDC Connector 2.4+ 版本，结合 DataStream API 实现实时数据同步。 --- #### **一、DataStream API 核心流程** ```java import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import com.ververica.cdc.connectors.mysql.source.MySqlSource; public class MySQLCDCExample { public static void main(String[] args) throws Exception { // 1. 创建 MySQL CDC 源 MySqlSource<String> source = MySqlSource.<String>builder() .hostname("localhost") .port(3306) .databaseList("order_db") // 监控的数据库 .tableList("order_db.orders") // 监控的表 .username("flink_user") .password("flink_pass") .deserializer(new JsonDebeziumDeserializationSchema()) // JSON 格式解析 .serverTimeZone("Asia/Shanghai") .serverId("5400-5405") // 唯一ID范围（避免冲突） .build(); // 2. 创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000); // 10秒Checkpoint（保证Exactly-Once） // 3. 构建处理链路 env.fromSource(source, WatermarkStrategy.noWatermarks(), "MySQL CDC Source") .map(json -> { // 自定义处理逻辑（如过滤/转换） System.out.println("Received: " + json); return json; }) .addSink(new FlinkKafkaProducer<>("topic", new SimpleStringSchema(), props)); env.execute("MySQL CDC Job"); } } ``` --- #### **二、关键配置说明** | 参数 | 作用 | 示例值 | |------|------|--------| | `serverId` | 唯一标识CDC连接 | `"5400-5405"` | | `scan.incremental.snapshot.enabled` | 增量快照（避免锁表） | `true` | | `scan.incremental.snapshot.chunk.size` | 分块读取大小 | `8096` | | `debezium.snapshot.mode` | 快照模式 | `initial`（全量+增量） | | `debezium.max.batch.size` | 单次读取事件数 | `2048` | --- #### **三、常见问题排查** 1. **无法连接 MySQL** - **原因**：权限不足或网络不通 - **解决**： ```sql GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flink_user'; ``` 验证网络：`telnet mysql_host 3306` 2. **全量阶段锁表阻塞写入** - **原因**：未启用增量快照（`scan.incremental.snapshot.enabled=false`） - **解决**：配置为 `true`，避免 `FLUSH TABLES WITH READ LOCK`[^4] 3. **数据延迟高** - **优化方向**： - 增加并行度：`env.setParallelism(4)` - 调大批处理大小：`.debeziumProperties("debezium.max.batch.size", "4096")` - 减少 Checkpoint 间隔：`env.enableCheckpointing(5000)` 4. **主键冲突导致写入失败** - **场景**：目标表主键与 MySQL 不一致 - **解决**：在 Map 阶段转换主键或忽略重复数据 ```java .filter(json -> !isDuplicate(json)) // 自定义去重逻辑 ``` --- #### **四、高级特性** 1. **Exactly-Once 保证** Flink CDC 通过 Checkpoint 记录 binlog 位置，故障时从断点恢复[^4]。需确保： - 启用 Checkpoint - Sink 支持两阶段提交（如 Kafka/Kudu） 2. **Schema 变更处理** 捕获 `ALTER TABLE` 事件（需自定义逻辑）： ```java deserializer = new JsonDebeziumDeserializationSchema() { @Override public void deserialize(SourceRecord record, Collector<String> out) { if (record.valueSchema().name().endsWith(".Envelope")) { Struct value = (Struct) record.value(); String op = value.getString("op"); // "c"=插入, "u"=更新, "d"=删除 // 解析 schema 变更 } } }; ``` --- #### **五、性能优化建议** - **资源分配**： 每个 Source 并行子任务需独立 MySQL 连接，合理设置 `parallelism`（不超过表分区数） - **反压处理**： 启用 Flink 反压监控，增加 Sink 吞吐量（如 Kafka 批量提交） - **内存调优**： 增大 TaskManager 堆内存，避免 `OutOfMemoryError` --- ###