Flink CDC中mysql cdc支持全量的时候并发读取,这个怎么配置呀？

多并行度就行了 ，此回答整理自钉群“Flink CDC 社区”

Flink CDC从2.0版本开始支持全量数据的并发读取，这项特性在2021年8月的更新中被引入。要实现MySQL的全量并发读取，首先需要添加Flink CDC MySQL连接器的依赖。如果你使用Maven，可以在pom.xml文件中添加如下依赖：

<groupId>com.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>2.5-SNAPSHOT</version>

注意这里的版本号应使用已发布的版本，对于snapshot版本的依赖需要本地自行编译。

然后，你可以使用Flink的DataStream API或Table API来读取全量数据。以下是一个使用DataStream API的例子：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

// 创建流执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 创建表执行环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 设置MySQL的CDC连接器相关参数
String url = "jdbc:mysql://localhost:3306/test?serverTimezone=UTC";
String user = "root";
String password = "password";
String database = "test";
String table = "my_table";

// 使用Flink SQL读取全量数据
DataStream<Row> result = tableEnv.toAppendStream(tableEnv.sqlQuery("SELECT * FROM " + database + "." + table), Row.class);

以上代码将连接到MySQL数据库，并将指定表的所有数据作为初始状态读取出来。请确保替换上述代码中的数据库连接信息为你自己的数据库信息。

在Flink CDC中，MySQL CDC支持全量并发读取的配置如下：

1. 首先，确保你已经添加了MySQL JDBC驱动和Debezium MySQL Connector的依赖。

2. 在Flink SQL中创建源表时，使用debezium-sql-connector作为连接器名称，并指定MySQL的连接信息、要监控的数据库和表等参数。例如：

CREATE TABLE my_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = 'password',
  'database-name' = 'mydb',
  'table-name' = 'mytable',
  'debezium-internal.offset-storage' = 'org.apache.flink.streaming.connectors.kafka.KafkaOffsetBackingStore',
  'debezium-internal.offset-storage.topic' = 'mydb-offsets',
  'debezium-internal.offset-storage.partition.group.id' = 'mydb-group',
  'scan.startup.mode' = 'latest-offset',
  'scan.startup.timeout' = '30000',
  'format' = 'debezium-json',
  'debezium-sql-connector.history.kafka.bootstrap.servers' = 'localhost:9092',
  'debezium-sql-connector.history.kafka.topic' = 'dbhistory.mydb',
  'debezium-sql-connector.history.kafka.groupId' = 'mydb-group',
  'debezium-sql-connector.snapshot.mode' = 'initial',
  'debezium-sql-connector.max.queued.messages' = '10000',
  'debezium-sql-connector.max.retry.attempts' = '16',
  'debezium-sql-connector.heartbeat.interval' = '10000',
  'debezium-sql-connector.max.allowed.packet.size' = '5242880',
  'checkpointing.interval' = '60000',
  'parallelism' = '4' -- 设置并行度为4
);

1. 然后，执行Flink SQL查询以消费数据。例如：

SELECT * FROM my_source;

通过以上配置，Flink CDC中的MySQL CDC应该可以支持全量并发读取。如果仍然遇到问题，请检查Flink集群的资源分配情况，确保有足够的资源来支持并发读取。

在Flink中，并行度的设置是在JobGraph层面进行的，而不是在Task层面进行的。JobGraph是Flink作业的静态描述，包含了作业的所有信息，包括输入输出、算子、并行度等。而Task是JobGraph在执行时的实例化，每个Task负责处理一部分数据。

在你的截图中，我们可以看到JobManager的Config显示了Parallelism为4，这意味着整个Flink作业的并行度被设置为4。然而，这并不意味着每个Task都会以4个并行度运行。在每个Task中，并行度是由算子的并行度决定的。

对于Flink CDC的MySQL连接器来说，它的并行度默认为1，也就是说，无论作业的并行度如何设置，Flink CDC的MySQL连接器只会启动一个并行任务来读取MySQL的binlog。这是因为Flink CDC的MySQL连接器需要保证数据的一致性，不能同时从同一个binlog position读取数据。

如果你想提高Flink CDC的MySQL连接器的并行度，你需要修改连接器的配置，将parallelism属性设置为大于1的值。例如，你可以通过编程的方式，在创建SourceFunction时设置parallelism属性，或者通过命令行的方式，在启动Flink任务时设置parallelism属性。