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，然后执行

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，然后执行

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，然后执行

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_show_sync_state()查询Hud同步任务状态：

1

SELECT * FROM hudi_show_sync_state();