Hudi任务同步
创建Hudi任务
迁移场景
如果GaussDB(DWS)表已经通过CDL导入数据,改为用SQL on Hudi方式迁移数据。或者使用CDM做全量初始化后,继续使用SQL on Hudi方式同步增量数据。
- 创建hudi.hudi_sync_state同步状态表,需要管理员权限。
1
SELECT pg_catalog.create_hudi_sync_table();
通常情况下,每个数据库中只创建一次hudi.hudi_sync_state。
- 设置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');
- 提交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; |
名称 |
类型 |
描述 |
---|---|---|
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'); |