DMS Kafka同步到Hudi作业配置
支持的源端和目的端数据库版本
源端数据库 |
目的端数据库 |
---|---|
Kafka集群(2.7、3.x版本) |
MRS集群(3.2.0-LTS.x、3.3.x-LTS) Hudi版本(0.11.0) |
数据库账号权限要求
在使用Migration进行同步时,源端和目的端所使用的数据库账号需要满足以下权限要求,才能启动实时同步任务。不同类型的同步任务,需要的账号权限也不同,详细可参考下表进行赋权。
类型名称 |
权限要求 |
---|---|
源数据库连接账号 |
Kafka开启密文接入场景下,所配置用户需要有发布和订阅Topic的权限,其余场景无特殊权限要求。 |
目标数据库连接账号 |
MRS用户需要拥有Hadoop和Hive组件的读写权限,建议参照图1所示角色及用户组配置MRS用户。
图1 MRS Hudi最小化权限
![]() 具体MRS集群角色权限管理请参考《MRS集群用户权限模型》。 |

- 建议创建单独用于Migration任务连接的数据库账号,避免因为数据库账号密码修改,导致的任务连接失败。
- 连接源和目标数据库的账号密码修改后,请同步修改管理中心对应的连接信息,避免任务连接失败后自动重试,导致数据库账号被锁定影响使用。
支持的同步对象范围
在使用Migration进行同步时,不同类型的链路,支持的同步对象范围不同,详细情况可参考下表。
类型名称 |
使用须知 |
---|---|
同步对象范围 |
支持同步所有Kafka消息,其中支持对JSON或CSV格式的消息体进行解析。 |
注意事项
除了数据源版本、连接账号权限及同步对象范围外,您还需要注意的事项请参见下表。
类型名称 |
使用和操作限制 |
---|---|
数据库限制 |
|
使用限制 |
通用:
增量同步阶段: 对于版本低于3.2.0-LTS1.5的MRS集群的Hudi MOR表,使用Migration(Flink)增量同步数据后,不支持直接使用CDM或Spark SQL写入数据,需要先进行Compaction后方可写入。 常见故障排查: 在任务创建、启动、全量同步、增量同步、结束等过程中,如有遇到问题,可先参考常见问题章节进行排查。 |
其他限制 |
- |
操作步骤
本小节以Kafka到MRS Hudi的实时同步为示例,介绍如何配置Migration实时集成作业。配置作业前请务必阅读使用前自检概览, 确认已做好所有准备工作。
- 参见新建实时集成作业创建一个实时集成作业并进入作业配置界面。
- 选择数据连接类型:源端选Kafka,目的端选Hudi。
图2 选择数据连接类型
- 选择集成作业类型:同步类型默认为实时,同步场景包含单表场景。
图3 选择集成作业类型
- 配置网络资源:选择已创建的DMS Kafka、MRS Hudi数据连接和已配置好网络连接的migration资源组。
图4 选择数据连接及migration资源组
无可选数据连接时,可单击“新建”跳转至管理中心数据连接界面,单击“创建数据连接”创建数据连接,详情请参见配置DataArts Studio数据连接参数进行配置。
无可选migration资源组时,可单击“新建”跳转至购买migration资源组页面创建migration资源组配置,详情请参见购买创建数据集成资源组增量包进行配置。
- 检测网络连通性:数据连接和migration资源组配置完成后需要测试整个迁移任务的网络连通性,可通过以下方式进行数据源和migration资源组之间的连通性测试。
- 单击展开“源端配置”触发连通性测试,会对整个迁移任务的连通性做校验。
- 单击源端和目的端数据源和migration资源组中的“测试”按钮进行检测。
- 配置源端参数。
- 主题
- 数据格式
- 消费组ID
消费者是从Topic订阅消息的一方,消费组是由一个或多个消费者组成的。Migration支持指定本次消费动作所属的Kafka消费组。
- Kafka源端属性配置
支持设置Kafka的配置项,需要增加 properties. 前缀,作业将自动移除前缀并传入底层Kafka客户端,具体参数可参考Kafka官方文档中的配置说明。
- 上游数据源类型
仅当数据格式选择DEBEZIUM_JSON或CANAL_JSON时需要配置,当前仅支持选择TiDB,标志CDC JSON为TiDB产生的。
- 库表名称列表
仅当数据格式选择DEBEZIUM_JSON或CANAL_JSON时需要配置,标志Kafka消息来源的数据库表,任务会解析CDC JSON并且只同步已配置库表的数据。
- 配置目的端参数。
图6 Hudi目的端配置项
- 目标表的基本配置
- 目标表名称
- 数据存储路径
Hudi自动建表时的warehouse路径,每张表会在warehouse路径下创建子目录。支持填写HDFS和OBS路径,路径格式参考:
- OBS路径:obs://bucket/warehouse。
- HDFS路径:/tmp/warehouse。
- Hudi表属性全局配置
支持通过参数配置部分高级功能,参数详情可参考Hudi高级配置一览表。
表5 Hudi高级配置一览表 参数名
参数类型
默认值
单位
参数说明
write.precombine.field
string
-
-
Hudi表的预聚合键,指在写入数据时用于合并记录的字段,通常用于处理更新和删除操作。本链路必须配置,请填写write.precombine.field属性并指定为Hudi表中的某一个字段。
index.type
string
BLOOM
-
Hudi表索引类型。
支持BLOOM和BUCKET索引,数据量较大场景下强烈建议使用BUCKET索引性能更好。
hoodie.bucket.index.num.buckets
int
256
个
Hudi表单分区下Bucket桶数。
说明:使用Hudi BUCKET表时需要设置Bucket桶数,桶数设置关系到表的性能,需要格外引起注意。
- 非分区表桶数 = MAX(CEIL(单表数据量大小(GB)/1GB), 4)。
- 分区表桶数 = MAX(CEIL(单分区数据量大小(GB)/1GB), 1)。
其中,要注意的是:
- 需要使用的是表的总数据大小,而不是压缩以后的文件大小。
- 桶的设置以偶数最佳,非分区表最小桶数请设置4个,分区表最小桶数请设置1个。
changelog.enabled
boolean
false
-
Hudi changelog功能开关,开启后Migration作业可输出DELETE和UPDATE BEFORE数据。
logical.delete.enabled
boolean
true
-
逻辑删除开关,changelog开启时必须关闭逻辑删除。
hoodie.write.liststatus.optimized
boolean
true
-
写log文件时是否开启liststatus优化。涉及到大表和分区数据量多的作业,在启动时list会非常耗时,可能导致作业启动超时,建议关闭。
hoodie.index.liststatus.optimized
boolean
false
-
定位数据时是否开启liststatus优化。涉及到大表和分区数据量多的作业,在启动时list会非常耗时,可能导致作业启动超时,建议关闭。
compaction.async.enabled
boolean
true
-
异步compaction开关。compaction操作一定程度会影响实时任务的写入性能,如果用户使用外置的compaction操作对hudi进行compaction,可以考虑设置为false关闭实时处理集成作业的compaction操作。
compaction.schedule.enabled
boolean
true
-
生成compaction计划的开关。compaction计划必须由本服务生成,计划的执行可以交给Spark。
compaction.delta_commits
int
5
次
生成compaction request的频率。compaction request生成频率降低可以使得compaction频率降低从而提升作业性能。如果hudi增量数据较小。可以考虑增大该值。
说明:例如配置为40,即每40次commit生成一个compaction request,因为Migration每分钟生成1个commit,那么每个compaction request将间隔40分钟。
clean.async.enabled
boolean
true
-
做历史版本数据文件清理的开关。
clean.retain_commits
int
30
次
要保留的commit数。这些commit关联的数据文件版本将被保留 num_of_commits * time_between_commits 这么长的时间,建议配置为2倍的compaction.delta_commits。
说明:例如配置为80,因为Migration每分钟生成1个commit,那么超过80分钟后如果有旧版本数据文件 ,则会生成clean request,且在执行clean时保留最近80个commit。
hoodie.archive.automatic
boolean
true
-
Hudi commit文件老化开关。
archive.min_commits
int
40
次
将旧版commit归档到日志文件中时要保留不归档的最小commit数。建议配置成clean.retain_commits + 1。
说明:例如配置成81,那么在触发归档动作时,将会保留最近81次commit文件。
archive.max_commits
int
50
次
触发归档动作的commit数。建议配置成archive.min_commits + 20。
说明:例如配置成101,那么将在生成101个commit文件后触发归档commit文件动作。
- 为了达到Migration作业性能最优,建议使用Hudi Bucket索引的MOR表,并根据实际数据量配置Bucket桶数。
- 为了保证Migration作业的稳定性,建议将Hudi Compaction单独拆成Spark作业交由MRS执行,在Migration任务里仅开启生成compaction计划,具体可以参考如何配置Hudi Compaction的Spark周期任务?。
- 刷新源表和目标表映射,检查映射关系是否正确,同时可根据需求修改表属性、添加附加字段,并通过“自动建表”能力在目的端Hudi数据库中建出相应的表。
源端为CDC JSON(DEBEZIUM_JSON和CANAL_JSON)数据时暂不支持自动建表,用户需手动创建Hudi表。图7 源表与目标表映射
- 同步主键
- 分区字段
目的端Hudi表未创建的情况下,可从映射出的目标字段中选择最多五个字段作为分区字段,Migration将在建表时自动指定Hudi表分区,同时写入Hudi时自动生成对应分区目录。
字段选择顺序影响分区的层级,例如选择par1、par2作为分区字段,那么par1为一级分区,par2为二级分区,最多支持五级分区。
- 目标字段编辑
Migration会根据选择的源端消息格式自动解析源端消息,生成对应的字段信息,用户可在此基础上进行编辑,自定义字段名、选择字段类型、填写字段值。
- 字段名称:目的端Hudi表新增字段的名称。
- 字段类型:目的端Hudi表新增字段的类型。
- 字段值:目的端Hudi表新增字段的取值来源。
表6 目标字段取值方式 类型
示例
手动赋值
任意字符。
内置变量
Kafka的元数据,包括__key__、__value__、__Topic__、__partition__、__offset__、__timestamp__共6个字段。
字段变量
从源端Kafka Topic消息解析出的任一字段。
udf方法
支持填写Flink的内置函数用于做数据转换,例如:
- CONCAT(CAST(NOW() as STRING), `col_name`)
- DATE_FORMAT(NOW(), 'yy')
其中的字段名要用反引号包围起来。Flink完整内置函数可参考Flink官方文档。
- 配置任务属性。
表7 任务配置参数说明 参数
说明
默认值
执行内存
作业执行分配内存,跟随处理器核数变化而自动变化。
8GB
处理器核数
范围:2-32。
每增加1处理核数,则自动增加4G执行内存和1并发数。
2
并发数
作业执行支持并发数。该参数无需配置,跟随处理器核数变化而自动变化。
1
自动重试
作业失败时是否开启自动重试。
否
最大重试次数
“自动重试”为是时显示该参数。
1
重试间隔时间
“自动重试”为是时显示该参数。
120秒
是否写入脏数据
选择是否记录脏数据,默认不记录脏数据,当脏数据过多时,会影响同步任务的整体同步速度。
- 否:默认为否,不记录脏数据。
- 是:允许脏数据,即任务产生脏数据时不影响任务执行。
允许脏数据并设置其阈值时:
- 若产生的脏数据在阈值范围内,同步任务将忽略脏数据(即不会写入目标端),并正常执行。
- 若产生的脏数据超出阈值范围,同步任务将失败退出。
说明:
脏数据认定标准:脏数据是对业务没有意义,格式非法或者同步过程中出现问题的数据;单条数据写入目标数据源过程中发生了异常,则此条数据为脏数据。 因此只要是写入失败的数据均被归类于脏数据。
例如,源端是VARCHAR类型的数据写到INT类型的目标列中,则会因为转换不合理导致脏数据不会成功写入目的端。用户可以在同步任务配置时,配置同步过程中是否写入脏数据,配置脏数据条数(单个分片的最大错误记录数)保证任务运行,即当脏数据超过指定条数时,任务失败退出。
否
脏数据策略
“是否写入脏数据”为是时显示该参数,当前支持以下策略:
- 不归档:不对脏数据进行存储,仅记录到任务日志中。
- 归档到OBS:将脏数据存储到OBS中,并打印到任务日志中。
不归档
脏数据写入连接
“脏数据策略”选择归档到OBS时显示该参数。
脏数据要写入的连接,目前只支持写入到OBS连接。
-
脏数据目录
脏数据写入的OBS目录。
-
脏数据阈值
是否写入脏数据为是时显示该参数。
用户根据实际设置脏数据阈值。
说明:- 脏数据阈值仅针对每个并发生效。比如阈值为100,并发为3,则该作业可容忍的脏数据条数最多为300。
- 输入-1表示不限制脏数据条数。
100
添加自定义属性
支持通过自定义属性修改部分作业参数及开启部分高级功能,详情可参见任务性能调优章节。
-
- 提交并运行任务。
作业配置完毕后,单击作业开发页面左上角“提交”,完成作业提交。
图8 提交作业提交成功后,单击作业开发页面“启动”按钮,在弹出的启动配置对话框按照实际情况配置同步位点参数,单击“确定”启动作业。
图9 启动配置表8 启动配置参数 参数
说明
偏移量参数
- 最早:从Kafka Topic最早偏移量开始消费数据。
- 最新:从Kafka Topic最新偏移量开始消费数据。
- 起止时间:根据时间获取Kafka Topic对应的偏移量,并从该偏移量开始消费数据。
起始时间
起止时间需要设置该参数,指示同步起始的时间位点。
说明:配置的位点时间早于Kafka消息最早偏移量时,默认会从最早偏移量开始消费。
- 监控作业。
通过单击作业开发页面导航栏的“前往监控”按钮,可前往作业监控页面查看运行情况、监控日志等信息,并配置对应的告警规则,详情请参见实时集成任务运维。
图10 前往监控
性能调优
若链路同步速度过慢,可参考参见任务性能调优章节中对应链路文档进行排查及处理。