更新时间:2024-12-06 GMT+08:00
分享

Hudi任务同步

创建Hudi任务

迁移场景

如果GaussDB(DWS)表已经通过CDL导入数据,改为用SQL on Hudi方式迁移数据。或者使用CDM做全量初始化后,继续使用SQL on Hudi方式同步增量数据。

  1. 创建hudi.hudi_sync_state同步状态表,需要管理员权限。

    1
    SELECT pg_catalog.create_hudi_sync_table();
    

    通常情况下,每个数据库中只创建一次hudi.hudi_sync_state。

  2. 设置CDL的同步进度,用户需要有同步目标表的INSERT和UPDATE权限、HUDI外表的SELECT权限,否则无法正常设置同步进度。

    1
    SELECT hudi_set_sync_commit('SCHEMA.TABLE', 'SCHEMA.FOREIGN_TABLE', 'LATEST_COMMIT');
    

    其中:

    • SCHEMA.TABLE,表示同步数据的目标表名,带schema。
    • SCHEMA.FOREIGN_TABLE,表示OBS外表命名,带schema。
    • LATEST_COMMIT,表示已同步的Hudi数据时间截点。

    示例:目标表public.in_rel,已经同步hudi的数据到20220913152131,切换到SQL on Hudi方式从OBS外表hudi_read1中继续导出数据。

    1
    SELECT hudi_set_sync_commit('public.in_rel', 'public.hudi_read1', '20220913152131');
    

  3. 提交Hudi同步任务。

    1
    SELECT hudi_sync_task_submit('SCHEMA.TABLE', 'SCHEMA.FOREIGN_TABLE');
    

    例如:目标表public.in_rel,切换到SQL on Hudi方式从OBS外表hudi_read1中继续导出数据。

    1
    SELECT hudi_sync_task_submit('public.in_rel', 'public.hudi_read1');
    

新建场景

DWS表为空,第一次从Hudi同步数据,可执行以下命令直接创建任务。

1
SELECT hudi_sync_task_submit('SCHEMA.TABLE', 'SCHEMA.FOREIGN_TABLE');

查询Hudi同步任务

查询Hudi同步任务,查询结果中的task_id是Hudi同步任务唯一标识。

1
SELECT * FROM pg_task_show('SQLonHudi');

暂停Hudi同步任务

查询Hudi任务,获取task_id暂停Hudi任务。

1
SELECT pg_task_pause('task_id');

示例:

暂停task_id为64479410-a04c-0700-d150-3037d700fffe的同步任务。

1
SELECT pg_task_pause('64479410-a04c-0700-d150-3037d700fffe');

恢复Hudi同步任务

查询Hudi任务,获取task_id,恢复Hudi任务。

1
SELECT pg_task_resume('task_id');

示例:

恢复task_id为64479410-a04c-0700-d150-3037d700fffe的同步任务。

1
SELECT pg_task_resume('64479410-a04c-0700-d150-3037d700fffe');

删除Hudi同步任务

查询Hudi任务,获取task_id,删除Hudi同步任务。

1
SELECT pg_task_remove('task_id');

示例:

删除task_id为64479410-a04c-0700-d150-3037d700fffe的同步任务。

1
SELECT pg_task_remove('64479410-a04c-0700-d150-3037d700fffe');

查询历史同步信息

使用视图hudi_sync_state_history_view查询Hudi历史同步任务信息,该视图仅9.1.0及以上集群版本支持。

1
SELECT * FROM pg_catalog.hudi_sync_state_history_view;
表1 hudi_sync_state_history_view字段

名称

类型

描述

task_id

TEXT

任务ID。

target_tbl

TEXT

同步目标表名。

source_ftbl

TEXT

同步源表名(外表)。

latest_commit

TEXT

最近一次同步成功的时间戳。

latest_sync_count

BIGINT

最近一次同步成功的行数。

latest_sync_start

TIMESTAMP WITH TIME ZONE

最近一次同步任务开始的时间。

latest_sync_end

TIMESTAMP WITH TIME ZONE

最近一次同步任务结束的时间。

hudi_flushdisk_time

TEXT

hudi文件落盘时间。

查询同步任务状态

使用函数hudi_show_sync_state()查询Hud同步任务状态:

1
SELECT * FROM hudi_show_sync_state();

复位连续失败Hudi同步任务

使用函数pg_task_resume()复位连续失败Hudi同步任务。

连续失败次数>=10的任务会自动暂停,需手动调用pg_task_resume()函数复位,该函数仅9.1.0及以上集群版本支持:

入参:连续失败Hudi任务的task_id。

1
SELECT pg_task_resume('task_id');

相关文档