谢谢jark！这几天一直在做性能调优！
1、这里针对这个简单场景目前可以在sink表的test_status表的primary key，增加一个join
key。即id和status两个列作为key，这样能使用数据最终一致，算是做了下规避，能一致。复杂点的语句感觉有点难搞，有点不敢用，主要不清楚这个乱序会对其它算子有什么影响，很容易出错，确实应该在flink框架里搞了合适些。这里jark在使用flink
sql cdc方面有啥建议吗？
2、关于性能这块，确实flink的rocksdb默认参数，性能很差！
Btw, I created an issue to track this problem:
https://issues.apache.org/jira/browse/FLINK-20374
Hope we can fix it in the next versions to have a better out-of-box
experience.
Best,
Jark
On Thu, 19 Nov 2020 at 13:58, Jark Wu wrote:
> 如果数据本身没什么倾斜，且并发也能打上去。那在 sql 这边也没什么其他办法了。得从 rocksdb
如果数据本身没什么倾斜，且并发也能打上去。那在 sql 这边也没什么其他办法了。得从 rocksdb 的角度去调优看看。比如：
1. 是否有使用 SSD?
2. 调整 write buffer 和 block cache
3. 更多可以看下这些 state 调优文章[1][2].
Best,
Jark
[1]: https://mp.weixin.qq.com/s/r0iPPGWceWkT1OeBJjvJGg
[2]: https://mp.weixin.qq.com/s/YpDi3BV8Me3Ay4hzc0nPQA
On Thu, 19 Nov 2020 at 12:19,
很感谢jark！
1、昨天将status表设置成时态表（Temporal
Tables），然后连续join试了下。确实有你说的问题，status表的更新不会触发任务计算，所有的数据实时变更需要test流来驱动。
同时时态表TTL设置问题，太小i/o有问题，太大结果不及时，与应用场景要求不符合，主要我们的场景下，status表也并不是维表，并且也数据量也大，变化也多。
2、并发度1的话，可以预见的是有主要性能问题，表大的情况下，join导致的反压厉害。
3、因为多并发度（10，20，40，80）测试中，我将join的两个表（test,
我再仔细看了下你的问题，你的 join key 是 status id，所以目前会按照 status id 做 shuffle key 分发给
join 的不同并发处理。
如果 test 表的 status id 发生变更的话，就会导致一个 test id 的数据会被不同的 join 并发处理，也即 test
数据已经乱序了，
这时候，即使下游再加 keyby sink key，也无济于事了。
所以，如果双流 join 两个 cdc 流，要注意 join key 是不能发生变更的，否则只能 join 设置成单并发。
像你这个场景，可以考虑采用维表 join status
1、试了下
在test表中增加一个proctime
CREATE TABLE test (
`id` INT,
`name` VARCHAR(255),
`time` TIMESTAMP(3),
`status` INT,
`proctime` AS PROCTIME(),
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '3306',
'username' = 'no_lock',
'password' =
1.
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/queries.html#deduplication
2. 这是1.12 的功能，定义在 sink DDL with 属性里的。
On Mon, 16 Nov 2020 at 14:18, jindy_liu <286729...@qq.com> wrote:
> 哦，这样啊
> 1、加上一个 deduplicate by sink key 节点在sql中是怎么写的？
> 2、另外sql 中有关键字能单独指定一条sink
哦，这样啊
1、加上一个 deduplicate by sink key 节点在sql中是怎么写的？
2、另外sql 中有关键字能单独指定一条sink sql的并发度吗？
--
Sent from: http://apache-flink.147419.n8.nabble.com/
图片png格式，怕看不了，我文字补充下：
1、print的最后几行。
32> +I(191,jindy191,2020-07-03T18:04:22,0,statu0)
32> +I(192,jindy192,2020-07-03T18:04:22,0,statu0)
32> +I(193,jindy193,2020-07-03T18:04:22,0,statu0)
32> +I(194,jindy194,2020-07-03T18:04:22,0,statu0)
32>
如果你是改了test表上的 status 关联字段，那么是会出现这个现象的。你一开始的 example 不是改 status 字段的。
这个问题的本质是 join key 和你最终的 sink key 不一致，导致可能出现乱序。
这个只需要在 sink 前显式按照 sink key shuffle 应该就能解决，比如加上一个 deduplicate by sink key 节点。
或者在 1.12 版本中，只需要 sink 并发与前面节点的并发不一样，框架也会自动加上一个 sink key shuffle。
关于你说的 join 节点热点问题，那是因为你的 status key
怕图片看不清，
我文字补充下：
1、print的最后几行。
32> +I(191,jindy191,2020-07-03T18:04:22,0,statu0)
32> +I(192,jindy192,2020-07-03T18:04:22,0,statu0)
32> +I(193,jindy193,2020-07-03T18:04:22,0,statu0)
32> +I(194,jindy194,2020-07-03T18:04:22,0,statu0)
32>
图片是屏幕截图，png格式的。忘记加后缀了。
--
Sent from: http://apache-flink.147419.n8.nabble.com/
我又重试了次，不用重启job也会有问题，就是把并行度大于1会有问题！。
1、直接在sql-client里，启动/data/home/jindyliu/flink-demo/flink-1.11.2/bin/sql-client.sh
embedded -d
/data/home/jindyliu/flink-demo/flink-1.11.2//conf/sql-client-defaults.yaml
sql-client-defaults.yaml的并行度设置为40.
数据一样，其中test表规模是200w条，status表11条。
源表test：
CREATE TABLE
面的任务不同并行度下，1和大于1的情况下，在并行度大于1的情况下，结果跟预期不相符。
>> /data/home/jindyliu/flink-demo/flink-1.11.2/bin/flink -s savepoint -p 1
>> job 下，
>> ==> test_status中的数据正常：
>> 0, name0, 2020-07-06 00:00:00 , 0, status0
>> 1, name1, 2020-07-06 00:00:00 , 1, status1_modify
>> 2, name2, 2020-07-06 00:00:00 , 1, status1_modify
>> /data/home/jindyliu/flink-demo/flink-1.11.2/bin/flink -s savepoint -p 40
>> job 下
>> ==> test_status中的数据不正常， id = 1,2的两条数据缺失：
>> 0, name0, 2020-07-06 00:00:00 , 0, status0
>>
>>
>> 怀疑与cdc的变化流在多个并行度下sink输出流的时候，先执行了 insert id=1再执行delete id=1的动作，导致数据有问题！！！
>>
>> 这里是不是bug？还是从save point里恢复的时候，算子的状态有问题？
>> 如果是，能不能在sink的时候，只把sink这里的并行度设置为1？？
>>
>>
>>
>>
>>
>>
>>
>> --
>> Sent from: http://apache-flink.147419.n8.nabble.com/
>>
>
00:00:00 , 1, status1_modify
> /data/home/jindyliu/flink-demo/flink-1.11.2/bin/flink -s savepoint -p 40
> job 下
> ==> test_status中的数据不正常， id = 1,2的两条数据缺失：
> 0, name0, 2020-07-06 00:00:00 , 0, status0
>
>
> 怀疑与cdc的变化流在多个并行度下sink输出流的时候，先执行了 insert id=1再执行delete id=1的动作，导致数据有问题！！！
>
> 这里是不是bug？还是从save point里恢复的时候，算子的状态有问题？
> 如果是，能不能在sink的时候，只把sink这里的并行度设置为1？？
>
>
>
>
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
>
/flink -s savepoint -p 40
job 下
==> test_status中的数据不正常， id = 1,2的两条数据缺失：
0, name0, 2020-07-06 00:00:00 , 0, status0
怀疑与cdc的变化流在多个并行度下sink输出流的时候，先执行了 insert id=1再执行delete id=1的动作，导致数据有问题！！！
这里是不是bug？还是从save point里恢复的时候，算子的状态有问题？
如果是，能不能在sink的时候，只把sink这里的并行度设置为1？？
--
Sent f
备集群同步效率会更高
>
>
>
>
> -- 原始邮件 --
> 发件人: tison 发送时间: 2020年5月17日 20:50
> 收件人: user-zh 主题: 回复：save point容灾方案咨询
>
>
>
> 这个我理解不在 Flink 的范畴里啊。你 savepoint 存到一个位置，然后外部挂一个同步器在主集群和容灾集群里同步（savepoint
> 目录）就可以吧。
>
> Best,
> tison.
>
+1，如果主备都在flink内的话，可能会加倍做checkpoint的负载，个人理解直接在状态后端内部做主备集群同步效率会更高
-- 原始邮件 --
发件人: tison
<854194...@qq.com> 于2020年5月17日周日 下午7:32写道：
>
> > 谢谢关注：
> >
> >
> > savepoint 容灾 是指的，每次执行savepoint生成的文件，能够在容灾集群上做备份。当主集群变得不可用时，可以将任务迁移到容灾
> > 集群进行根据savepoint 进行任务恢复。
> >
> >
> > --原始邮件--
> > 发件人
群进行根据savepoint 进行任务恢复。
>
>
> --原始邮件--
> 发件人:"Congxian Qiu" 发送时间:2020年5月17日(星期天) 晚上6:01
> 收件人:"user-zh"
> 主题:Re: save point容灾方案咨询
>
>
>
> 你好
>
> 请问这里的 savepoint 容灾的 “容灾” 具体是指什么呢？希望解决什么问题呢？
>
> Best,
> Congxian
>
>
> La
??
savepoint
savepoint??
savepoint ??
----
??:"Congxian Qiu"
你好
请问这里的 savepoint 容灾的 “容灾” 具体是指什么呢？希望解决什么问题呢？
Best,
Congxian
LakeShen 于2020年5月15日周五 上午10:20写道：
> Hi ,
>
> 你可以把你的场景在描述的详细一些。
>
> Best,
> LakeShen
>
> 请叫我雷锋 <854194...@qq.com> 于2020年5月14日周四 下午9:42写道：
>
> > 各位大佬好，请问有啥好的save point容灾方案嘛？
> >
> >
> >
> > 发自我的iPhone
>
Hi ,
你可以把你的场景在描述的详细一些。
Best,
LakeShen
请叫我雷锋 <854194...@qq.com> 于2020年5月14日周四 下午9:42写道：
> 各位大佬好，请问有啥好的save point容灾方案嘛？
>
>
>
> 发自我的iPhone
各位大佬好，请问有啥好的save point容灾方案嘛？
发自我的iPhone
24 matches
Mail list logo
