流批一体的近实时数仓的思考与设计

两张表：dwd_partition_word_count，dws_partition_word_count，计算 word count

CREATE TABLE tablestore.tablestore_test.dwd_partition_word_count (  logdate String,    user_id bigint) PARTITIONED BY (logdate)WITH (    'bucket' = '3');
CREATE TABLE tablestore.tablestore_test.dws_partition_word_count (  logdate String,    user_id bigint,    cnt BIGINT,    PRIMARY KEY (logdate,user_id) NOT ENFORCED) PARTITIONED BY (logdate)WITH (    'bucket' = '3');


批任务的 Flink SQL：

insert overwrite tablestore.tablestore_test.dws_partition_word_count PARTITION(logdate=${start_date}) select user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count where logdate=${start_date} group by user_id;-- 或者insert overwrite tablestore.tablestore_test.dws_partition_word_countselect logdate, user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count where logdate=${start_date} group by logdate,user_id;


流任务的 Flink SQL:


insert into tablestore.tablestore_test.dws_partition_word_count select logdate,user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count group by logdate,user_id;


如何用一个 Flink SQL 来实现流批模型下的不同呢？


不同点：Insert into 和 Insert overwrite 的问题，这个通过在提交运行模式的时候，如果是批任务，则是 Insert Overwrite，如果是流任务，则转为 Insert into，这个在技术上没有什么难点。


不同点：Where 条件的数据范围问题。抽象来看，流任务和批任务的时间范围在表达上是可以统一的


insert overwrite tablestore.tablestore_test.dws_partition_word_countselect logdate, user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count where logdate>=${start_date} and logdategroup by logdate,user_id;


比如跑 4 月 22 号一天的数据，执行的批 SQL 为：

insert overwrite tablestore.tablestore_test.dws_partition_word_countselect logdate, user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count where logdate>='20230422' and logdate'20230422' group by logdate,user_id;


如果用流模式跑，执行的 SQL 可以为：



select logdate, user_id,count(1) as cnt from tablestore.tablestore_test.dwd_partition_word_count where logdate>='19700101' and logdate'99990101' group by logdate,user_id;




insert overwrite/into 和时间范围，可以由平台执行的时候自动转换和参数输入。


3. 批任务的调度和流任务的计算如何分离


任务完成开发，在批模式下，用调度任务验证了逻辑无误，那么之后可以用流模式，一直持续不断的运行。一是计算逻辑变更或者历史数据修复怎么办，二是可不可以支持流批双跑。其实本质是一个问题。如果计算逻辑变更，那么可以修改流批一体的 SQL 逻辑，然后流任务重启应用新的计算逻辑。同时，流批一体的 SQL，在调度上回跑历史数据，重新刷写数据。


重刷历史数据的时候，流任务会不会读取到重刷的历史数据进行计算。


这个问题主要是通过上述说的数据湖版本 commit kind 解决。批任务只应用 insert overwrite，流任务应用 insert into.如果流任务检测到 insert overwrite 的版本提交，直接跳过，不做实际的数据读取和处理。只处理 insert into  的数据。实际批任务的执行，对流任务不会产生影响。


目前在数据湖流式读取上，只需要加个开关选项就可以实现。


4. 流任务的 Insert into 如何实现主键写入


如果流任务的 Insert into 不能实现主键写入，那么分区数据的重复性无法解决，那么就只能流批双跑来解决数据的重复性问题。也就是，下游如果是主键幂等写入，insert into 和 insert overwrite 语义等同。


这个可以通过数据湖主键表(比如 Paimon 的主键表)实现。Paimon 的主键表已初步具备生产可用性。


5. 流批任务的调度依赖

如果一个流任务，下游接的是批任务调度，如果实现调度依赖呢？


比较优雅的实现可以是，在流任务写入下游表的时候，假如数据的 Watermark 写入到下游表的属性中，如果最晚的数据已经是当前小时的 05 分，那么当前小时的下游调度任务，通过检查表的属性时间，就可以判断批任务的调度实例是否应该拉起。或者也可以基于流任务的运行延迟做检查依赖。


基于上述的实现和解决，我们基本就可以实现流批一体的 Flink SQL 在批模式和流模式下运行，如果调度依赖做的比较完善的情况下，可以实现流批混跑。同时补数或者双跑对流任务的稳定性不会产生影响。


实际开发，就可以用批任务先开发验证，然后用流模式拉起，数据产出基本是分钟级别的。出问题可以用批任务修正。



            

            

            

                

                    0
                    0
                    投票数
                
文章评分
                

            
                                                        
本文转载自张剑 Apache Flink，原文链接：https://mp.weixin.qq.com/s/Pn_f01__R9IoqnkR_C59ow。