贝壳找房基于 Flink+Paimon 进行全量数据实时分组排序的实践

select project_id  -- 项目id    ,contract_id -- 合同id    ,sign_time  -- 签约时间    ,FROM_UNIXTIME(UNIX_TIMESTAMP(sign_time,'yyyy-MM-dd HH:mm:ss')/(24*60*60)*(24*60*60)           ,'yyyy-MM-dd HH:mm:ss'           ) as event_time_group_start_point -- 事件时间分段起点 from default_catalog.default_database.mysql_project_contract_table -- 项目合同信息

insert into paimon_catalog.paimon_db.dwd_project_contract_memory_result_table -- 实时数据排序结果​select project_id  -- 项目id          ,contract_id -- 合同id      ,sign_time  -- 签约时间             ,event_time           ,event_time_group_start_point -- 事件时间分段起点`    ,DENSE_RANK() over (PARTITION BY project_id ,event_time_group_start_point ORDER BY event_time asc) as mem_rn -- 实时数据排序结果 from paimon_catalog.paimon_db.dwd_project_contract_table -- 项目合同信息`

-- 分组内聚合的累计结果CREATE TABLE if not exists paimon_catalog.paimon_db.dwd_project_contract_final_result_max_table (   project_id   bigint COMMENT '项目id'  ,max_rn     bigint COMMENT '该项目下当前最大的rn值'  ,max_event_time string  COMMENT '该项目下当前最大的event_time'    ,PRIMARY KEY (project_id) NOT ENFORCED) WITH (   'merge-engine' = 'aggregation'  ,'changelog-producer' = 'lookup'  ,'fields.max_rn.aggregate-function' = 'max'  ,'fields.max_rn.ignore-retract'='true'  ,'fields.max_event_time.aggregate-function' = 'max'  ,'fields.max_event_time.ignore-retract'='true');​insert into paimon_catalog.paimon_db.dwd_project_contract_final_result_max_table  ​select project_id -- 项目id           ,asc_rn   -- 分组排序结果    ,sign_time  -- 合同签约时间(事件时间) from paimon_catalog.paimon_db.dwd_project_contract_final_result_table -- 分组排序结果

CREATE TABLE if not exists paimon_catalog.paimon_db.dwd_project_contract_final_result_table_max_history_table (   project_id      bigint COMMENT '项目id'  ,pre_max_rn      bigint COMMENT '该项目下当前最大的rn值'  ,pre_max_event_time  string COMMENT '该项目下当前最大的event_time'    ,next_max_event_time string COMMENT '被更新后该项目下当前最大的event_time'    ,PRIMARY KEY (project_id) NOT ENFORCED)WITH (   'merge-engine' = 'aggregation'  ,'changelog-producer' = 'lookup'  ,'fields.pre_max_rn.aggregate-function' = 'max'  ,'fields.pre_max_rn.ignore-retract'='true'  ,'fields.pre_max_event_time.aggregate-function' = 'max'  ,'fields.pre_max_event_time.ignore-retract'='true'  ,'fields.next_max_event_time.aggregate-function' = 'max'  ,'fields.next_max_event_time.ignore-retract'='true');​INSERT INTO paimon_catalog.paimon_db.dwd_project_contract_final_result_table_max_history_tableselect project_id       ,max_rn         ,max_event_time       ,'0000-00-00 00:00:00' from paimon_catalog.paimon_db.dwd_project_contract_final_result_max_table$audit_log where rowkind = '-U'​ union all​select project_id       ,0    ,'0000-00-00 00:00:00'        ,max_event_time   from paimon_catalog.paimon_db.dwd_project_contract_final_result_max_table$audit_log where rowkind = '+U'

CREATE TABLE if not exists paimon_catalog.paimon_db.dwd_project_contract_base_version_table (   project_id      bigint COMMENT '项目id'  ,base_ver_start_time  string  COMMENT '该版本开始时间'  ,base_ver_end_time   string  COMMENT '该版本结束时间'  ,base_ver_max_rn    bigint COMMENT '该版本最大的rn'  ,PRIMARY KEY (project_id,base_ver_start_time) NOT ENFORCED)WITH (   'merge-engine' = 'deduplicate'  ,'changelog-producer' = 'lookup');​INSERT INTO paimon_catalog.paimon_db.dwd_project_contract_base_version_table -- 累计结果拉链表select project_id          ,pre_max_event_time  as base_ver_start_time    ,next_max_event_time as base_ver_end_time    ,pre_max_rn      as base_ver_max_rn   from paimon_catalog.paimon_db.dwd_project_contract_final_result_max_history_table$audit_log -- 累计结果的版本变更信息的 audit log where rowkind in ('+I','+U')  and pre_max_event_time <> next_max_event_time​ union all ​select project_id       ,max_event_time     as base_ver_start_time    ,'9999-99-99 99:99:99' as base_ver_end_time       ,max_rn         as base_ver_max_rn from paimon_catalog.paimon_db.dwd_project_contract_final_result_max_table -- 累计结果

CREATE TABLE if not exists paimon_catalog.paimon_db.dwd_project_contract_final_result_table (   project_id           bigint    COMMENT '项目id'  ,contract_id          bigint    COMMENT '合同id'  ,sign_time           string    COMMENT '合同签约时间'  ,event_time           TIMESTAMP(3) COMMENT 'sign_time作为event_time'  ,event_time_group_start_point  STRING    COMMENT '业务时间分段起点'  ,mem_rn             bigint    COMMENT '这条数据在内存排序中排出来的rn'  ,base_ver_start_time      string    COMMENT '该版本开始时间'  ,base_ver_end_time       string    COMMENT '该版本结束时间'  ,base_ver_max_rn        bigint    COMMENT '该版本最大的rn'  ,asc_rn             bigint    COMMENT '最终排序结果'  ,PRIMARY KEY (project_id,contract_id,sign_time,event_time,event_time_group_start_point) NOT ENFORCED)WITH (   'merge-engine' = 'aggregation'  ,'changelog-producer' = 'lookup'  ,'fields.mem_rn.aggregate-function' = 'max'  ,'fields.mem_rn.ignore-retract'='true'  ,'fields.base_ver_start_time.ignore-retract'='true'  ,'fields.base_ver_end_time.ignore-retract'='true'    ,'fields.base_ver_max_rn.aggregate-function' = 'max'    ,'fields.base_ver_max_rn.ignore-retract' = 'true'      ,'fields.asc_rn.aggregate-function' = 'max'    ,'fields.asc_rn.ignore-retract'='true'         );​insert into paimon_catalog.paimon_db.dwd_project_contract_final_result_table​-- 实时数据 lookup join 累计结果的历史快照select s.project_id      ,s.contract_id    ,s.sign_time      ,s.event_time    ,s.event_step_start_point    ,s.mem_rn    ,coalesce(base.base_ver_start_time ,'0000-00-00 00:00:00') as base_ver_start_time    ,coalesce(base.base_ver_end_time  ,'0000-00-00 00:00:00') as base_ver_end_time      ,coalesce(base.base_ver_max_rn   ,0) as base_ver_max_rn       ,s.mem_rn + coalesce(base.base_ver_max_rn,0) as asc_rn from paimon_catalog.paimon_db.dwd_project_contract_memory_result_table s -- 实时数据的内存排序结果 left join paimon_catalog.paimon_db.dwd_project_contract_base_version_table FOR SYSTEM_TIME AS OF s.proctime AS base -- 累计结果拉链表  on s.project_id = base.project_id  and s.event_step_start_point >= base.base_ver_start_time   and s.event_step_start_point < base.base_ver_end_time