谢谢jark！这几天一直在做性能调优！
1、这里针对这个简单场景目前可以在sink表的test_status表的primary key，增加一个join
key。即id和status两个列作为key，这样能使用数据最终一致，算是做了下规避，能一致。复杂点的语句感觉有点难搞，有点不敢用，主要不清楚这个乱序会对其它算子有什么影响，很容易出错，确实应该在flink框架里搞了合适些。这里jark在使用flink
sql cdc方面有啥建议吗？
2、关于性能这块，确实flink的rocksdb默认参数，性能很差！
按你给的文章，调了些参数，同时换了ssd硬盘后，write_buff
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, jind
很感谢jark！
1、昨天将status表设置成时态表（Temporal
Tables），然后连续join试了下。确实有你说的问题，status表的更新不会触发任务计算，所有的数据实时变更需要test流来驱动。
同时时态表TTL设置问题，太小i/o有问题，太大结果不及时，与应用场景要求不符合，主要我们的场景下，status表也并不是维表，并且也数据量也大，变化也多。
2、并发度1的话，可以预见的是有主要性能问题，表大的情况下，join导致的反压厉害。
3、因为多并发度（10，20，40，80）测试中，我将join的两个表（test,
status）的数据完全对称，没有倾斜问题（200w，
我再仔细看了下你的问题，你的 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' = 'n
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 sql的并发
哦，这样啊
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> +I(195,jindy195,202
如果你是改了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> +I(195,jindy195,2020-07-
图片是屏幕截图，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 te
抱歉... 题目没有看仔细，才发现你说的是 es sink，那和我上面说的 bug 不是一个问题。
不过从理论分析，不应该出现这个现象。
我在本地1.11分支上，用你给的数据和 sql，也没有复现你说的问题。
是不是 sql 给的不对？我看你 test_status 表的定义在 pk 之前少了一个逗号..
Best,
Jark
On Sat, 14 Nov 2020 at 17:48, Jark Wu wrote:
> 看起来是这个 jdbc sink bug 导致的 https://issues.apache.org/jira/browse/FLINK-19423
> 这个 bu
看起来是这个 jdbc sink bug 导致的 https://issues.apache.org/jira/browse/FLINK-19423
这个 bug 会导致删的时候，取的 pk 索引不对，所以可能导致 index 异常，或是删错数据。
这个bug 会在即将发布的 1.11.3 中修复。
Best,
Jark
On Fri, 13 Nov 2020 at 13:12, jindy_liu <286729...@qq.com> wrote:
> 源表test：
> CREATE TABLE test (
> `id` INT,
> `name` VARCHAR(255)
源表test：
CREATE TABLE test (
`id` INT,
`name` VARCHAR(255),
`time` TIMESTAMP(3),
`status` INT,
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '3306',
'username' = 'root',
'password' = '1',
'database-name' = 'ai_audio_lyric_task',
't
16 matches
Mail list logo
