大佬请教几个问题 about flink-connector-clickhouse

[v-qiunan]

@itinycheng

[itinycheng]

1. 假如 10 个并行度去写本地表。本地表有3个，一个并行度是去写一个表，还是轮询写这三个表？
2. 假如服务端重启了，可以重试连副本节点继续写入？

您好，

1. Operator分发策略是flink端的逻辑，sink端能操作的是设置并行度：sink.parallelism；另外，如果设置写本地表，一个sink operator实例会持有该distributed表下所有本地表的connection。

2. 当前不支持，写本地表的分片选择是写死的，取的好像是shard_num=1，具体可以看下这个类：ClickHouseConnectionProvider ，如果分片挂啦，重启job没办法恢复。

[v-qiunan]

@itinycheng
您好，
1. 按照你的说法，假如我有 100 并行度， 100 个本地表。一个本地表，每次都有 100 个并行度往这个本地表写？也就是会生成 100 个文件？合并的压力就比较大。我这样理解的对不对？（之前我还想改造成按照并行度写，但是写入策略是根据某个 key的话就行不通了）
2. 我发现实时写 ck 会遇到很多问题，你们写入的时候，批量写间隔/写大小一般设置多大？

[itinycheng]

@itinycheng 您好， 1. 按照你的说法，假如我有 100 并行度， 100 个本地表。一个本地表，每次都有 100 个并行度往这个本地表写？也就是会生成 100 个文件？合并的压力就比较大。我这样理解的对不对？（之前我还想改造成按照并行度写，但是写入策略是根据某个 key的话就行不通了） 2. 我发现实时写 ck 会遇到很多问题，你们写入的时候，批量写间隔/写大小一般设置多大？

您好：

1. 首先100个并行度对应100个sinkOperator实例，每个sinkOperator实例会初始化100个connection（对应100个local table），一个connection会写一台机器的本地表；数据如果提前keyBy的可以将数据提前分片下，sink端文件生成会减少；sink并行度设置小一点也会减少文件生成数；自己测试下就明白了，只是说可能不太好理解；
2. 我生产批量写时间配置一般在5s～60s之间，batch-size根据实际情况，在千～百万之间不等；

[v-qiunan]

@itinycheng
您好：
对于第一个问题，假如sink并行度减少，数据处理不过来，会造成反压。是不是可以优化成，我查询到 100 个本地表。我并行度是1连第一个节点。并行度是2写第二个节点，以此类推呢？这种方案有什么缺陷？
对于 ck 集群的某个节点重启的时候。写入失败。我再去重新连接重启这个节点的副本。这种改动有意义？因为目前遇到的问题是，ck经常有增加减少字段的需求。当执行 DDL 的时候，经常卡主。需要重启 ck 节点。重启的时候实时任务就会报错，重启。重启之后任务有一段时间的延迟。

[itinycheng]

@itinycheng 您好： 对于第一个问题，假如sink并行度减少，数据处理不过来，会造成反压。是不是可以优化成，我查询到 100 个本地表。我并行度是1连第一个节点。并行度是2写第二个节点，以此类推呢？这种方案有什么缺陷？ 对于 ck 集群的某个节点重启的时候。写入失败。我再去重新连接重启这个节点的副本。这种改动有意义？因为目前遇到的问题是，ck经常有增加减少字段的需求。当执行 DDL 的时候，经常卡主。需要重启 ck 节点。重启的时候实时任务就会报错，重启。重启之后任务有一段时间的延迟。

并行度问题：sink并行度和上游不同情况下，记得好像是必须有primary key的，数据分发是通过 hash(primary key value) % parallelism 来确定写哪个sinker的，可以关注下：CommonExecSink.createSinkTransformation，具体效果自己测试下看；另外你说这个感觉不太合理，写入哪个local表必须要根据分片的key来确定，不然数据不就乱了，另外你可以直接写分布式表use-local=false，让分布式表帮分发数据；

切换sink的replica问题：DDL时候应该是不允许写入操作的吧，connector在这段时间可操作空间不大；支持切换replica是有价值的，有合理方案就行，或是你可以根据实际业务需求定制一个job recovery/fallover的策略；

[v-qiunan]

@itinycheng 好的。非常感谢。
最近我用批任务运行。两个sink。一个是kafka 。一个是clickhouse。任务结束之后。kafka有数据。clickhouse 没有任何数据。任务也没有任何报错。大佬有遇到过吗？

[v-qiunan]

@itinycheng batch 任务报错了。异常没抛出

[itinycheng]

@itinycheng batch 任务报错了。异常没抛出

更新到最新代码，如果还有BUG，欢迎提Issue或PR；

[Leijingwen]

@itinycheng > 并行度问题：sink并行度和上游不同情况下，记得好像是必须有primary key的，数据分发是通过 hash(primary key value) % parallelism 来确定写哪个sinker的，可以关注下：CommonExecSink.createSinkTransformation，具体效果自己测试下看；另外你说这个感觉不太合理，写入哪个local表必须要根据分片的key来确定，不然数据不就乱了，另外你可以直接写分布式表use-local=false，让分布式表帮分发数据；

大佬, 您好, 请教个问题 :我把 'use-local' = 'false' 的时候程序会有报错-- Table engine Distributed doesn't support mutations. (NOT_IMPLEMENTED) (version 22.2.2.1) , 设置为 true的时候明显丢了很多数据 , 'sink.partition-strategy' = 'hash',
'sink.partition-key' = 'xx' . 参数是指分布式表中第四个参数, 还是指代本地表中的partition?

[itinycheng]

@Leijingwen
这个参数命名不太好，应该叫 sharding key，是用来计算一条数据写入的具体分片；

[Leijingwen]

@Leijingwen 这个参数命名不太好，应该叫 sharding key，是用来计算一条数据写入的具体分片；

ok

[Leijingwen]

1. 假如 10 个并行度去写本地表。本地表有3个，一个并行度是去写一个表，还是轮询写这三个表？
2. 假如服务端重启了，可以重试连副本节点继续写入？

您好，

1. Operator分发策略是flink端的逻辑，sink端能操作的是设置并行度：sink.parallelism；另外，如果设置写本地表，一个sink operator实例会持有该distributed表下所有本地表的connection。
2. 当前不支持，写本地表的分片选择是写死的，取的好像是shard_num=1，具体可以看下这个类：ClickHouseConnectionProvider ，如果分片挂啦，重启job没办法恢复。

您好,大佬 :当有shard_num=1的机器挂掉机器以后,flink-任务就报错了, 这个有什么好的方式解决吗? clickhouse集群的高可用感觉用不上了. 设置'use-local' = 'false'?

[itinycheng]

1. 假如 10 个并行度去写本地表。本地表有3个，一个并行度是去写一个表，还是轮询写这三个表？
2. 假如服务端重启了，可以重试连副本节点继续写入？

您好，

1. Operator分发策略是flink端的逻辑，sink端能操作的是设置并行度：sink.parallelism；另外，如果设置写本地表，一个sink operator实例会持有该distributed表下所有本地表的connection。
2. 当前不支持，写本地表的分片选择是写死的，取的好像是shard_num=1，具体可以看下这个类：ClickHouseConnectionProvider ，如果分片挂啦，重启job没办法恢复。

您好,大佬 :当有shard_num=1的机器挂掉机器以后,flink-任务就报错了, 这个有什么好的方式解决吗? clickhouse集群的高可用感觉用不上了. 设置'use-local' = 'false'?

@Leijingwen

• connector 1.15版本以后使用的BalancedClickhouseDataSource来创建connection，Job start/restart会随机connect一个可用的replica，这时应该会有些高可用的效果；git commit: da116e4 。
• use-local 作用只是为了直接读写local表用的，没有高可用的效果；

[Leijingwen]

1. 假如 10 个并行度去写本地表。本地表有3个，一个并行度是去写一个表，还是轮询写这三个表？
2. 假如服务端重启了，可以重试连副本节点继续写入？

您好，

1. Operator分发策略是flink端的逻辑，sink端能操作的是设置并行度：sink.parallelism；另外，如果设置写本地表，一个sink operator实例会持有该distributed表下所有本地表的connection。
2. 当前不支持，写本地表的分片选择是写死的，取的好像是shard_num=1，具体可以看下这个类：ClickHouseConnectionProvider ，如果分片挂啦，重启job没办法恢复。

您好,大佬 :当有shard_num=1的机器挂掉机器以后,flink-任务就报错了, 这个有什么好的方式解决吗? clickhouse集群的高可用感觉用不上了. 设置'use-local' = 'false'?

@Leijingwen

• connector 1.15版本以后使用的BalancedClickhouseDataSource来创建connection，Job start/restart会随机connect一个可用的replica，这时应该会有些高可用的效果；git commit: da116e4 。
• use-local 作用只是为了直接读写local表用的，没有高可用的效果；

好的,谢谢, 我试试