Flink CDC报错不支持 要解决的问题是数据出现反压 读取速度远远大于写入速度么？

在 Flink 1.13.6 的 SQL CDC 中，你可以通过使用 SET 命令来分别设置 Source 和 Sink 的并行度。直接将并行度参数写入到 WITH 参数中是不支持的。

下面是一个示例代码片段，演示了如何设置 Source 和 Sink 的并行度：

SET 'source.parallelism' = '2';
SET 'sink.parallelism' = '4';

CREATE TABLE source_table (
    -- 定义表结构和字段
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '<MySQL 主机名>',
    'port' = '<MySQL 端口号>',
    'database-name' = '<数据库名>',
    'table-name' = '<表名>'
);

CREATE TABLE sink_table (
    -- 定义表结构和字段
) WITH (
    'connector' = 'kafka',
    'topic' = '<Kafka 主题>',
    'properties.bootstrap.servers' = '<Kafka 服务器地址>'
);

在上述示例中，通过使用 SET 命令将 source.parallelism 设置为 2，并将 sink.parallelism 设置为 4。这样可以分别指定 Source 和 Sink 的并行度。

另外，如果你遇到数据出现反压的问题，读取速度大于写入速度，可以尝试以下方法来解决：

• 增加 Sink 的并行度：通过增加 Sink 的并行度来提高写入数据的并行处理能力，以匹配读取数据的速度。
• 调整读取数据的速率：通过调整 Source 的配置参数，如 scan.incremental.snapshot.chunk.size 和 scan.snapshot.fetch.size 等，来限制读取数据的速率。
• 调整 Flink 作业的资源配置：根据实际情况，增加 TaskManager 的数量、调整内存分配等，以提供足够的计算和资源能力。

需要注意的是，并行度的设置应该根据实际情况进行调整，以保持整个作业的平衡和稳定运行。

希望这些信息对你有所帮助！如果还有其他问题，请继续提问。

这不是Sql操作吗，不能写到with里面，在ddl语句创建之前可以写，你去看看mysql案例，里面有个set 并行度的，此回答整理自钉群“Flink CDC 社区”