Re: flink cdc 如何保证group agg结果正确性

[Jark Wu]

你的数据源中是不是没有历史全量数据，所以发现结果对不上？

一般建议同步全量+增量数据到 kafka 中，然后flink 从头消费这个 topic。
另外 mysql-cdc connector [1] 也提供了全量+增量读取的能力。

Best,
Jark

[1]:
https://github.com/ververica/flink-cdc-connectors/wiki/MySQL-CDC-Connector


On Mon, 30 Nov 2020 at 22:54, kandy.wang  wrote:

> insert into kudu.default_database.index_agg
> SELECT v_spu_id as spu_id,sum(leaving_num*vipshop_price) as
> leaving_price,DATE_FORMAT(LOCALTIMESTAMP,'-MM-dd HH:mm:ss')
> FROM  XX.XX.XX
> group by v_spu_id;
>
>
> XX.XX.XX 是通过自定义cdc
> format消费公司的cdc数据源，cdc数据源在kafka，数据只保留7天数据，都是增量消费，如何保证结果准确。
> 怎么做初始化，这个初始化，是把数据初始化到state里么？ 现在通过对数发现，数据量对不上。

flink cdc 如何保证group agg结果正确性

[kandy.wang]

insert into kudu.default_database.index_agg
SELECT v_spu_id as spu_id,sum(leaving_num*vipshop_price) as 
leaving_price,DATE_FORMAT(LOCALTIMESTAMP,'-MM-dd HH:mm:ss')
FROM  XX.XX.XX
group by v_spu_id;


XX.XX.XX 是通过自定义cdc format消费公司的cdc数据源，cdc数据源在kafka，数据只保留7天数据，都是增量消费，如何保证结果准确。
怎么做初始化，这个初始化，是把数据初始化到state里么？ 现在通过对数发现，数据量对不上。

flink cdc 如何保证group agg结果正确性

[kandy.wang]

insert into kudu.default_database.index_agg
SELECT v_spu_id as spu_id,sum(leaving_num*vipshop_price) as 
leaving_price,DATE_FORMAT(LOCALTIMESTAMP,'-MM-dd HH:mm:ss')
FROM  XX.XX.XX
group by v_spu_id;


XX.XX.XX 是通过自定义cdc format消费公司的cdc数据源，cdc数据源在kafka，数据只保留7天数据，都是增量消费，如何保证结果准确。
怎么做初始化，这个初始化，是把数据初始化到state里么？ 现在通过对数发现，数据量对不上。