文档首页/ 数据治理中心 DataArts Studio/ 用户指南/ 数据集成（实时作业）/ 实时集成作业使用教程/ DMS Kafka同步到Hudi作业配置
更新时间：2025-09-09 GMT+08:00
查看PDF
分享

DMS Kafka同步到Hudi作业配置

支持的源端和目的端数据库版本

表1 支持的数据库版本

源端数据库

目的端数据库

Kafka集群（2.7、3.x版本）

MRS集群（3.2.0-LTS.x、3.3.x-LTS）

Hudi版本（0.11.0）

数据库账号权限要求

在使用Migration进行同步时，源端和目的端所使用的数据库账号需要满足以下权限要求，才能启动实时同步任务。不同类型的同步任务，需要的账号权限也不同，详细可参考下表进行赋权。

表2 数据库账号权限

类型名称

权限要求

源数据库连接账号

Kafka开启密文接入场景下，所配置用户需要有发布和订阅Topic的权限，其余场景无特殊权限要求。

目标数据库连接账号

MRS用户需要拥有Hadoop和Hive组件的读写权限，建议参照图1所示角色及用户组配置MRS用户。

图1 MRS Hudi最小化权限

具体MRS集群角色权限管理请参考《MRS集群用户权限模型》。
  • 建议创建单独用于Migration任务连接的数据库账号，避免因为数据库账号密码修改，导致的任务连接失败。
  • 连接源和目标数据库的账号密码修改后，请同步修改管理中心对应的连接信息，避免任务连接失败后自动重试，导致数据库账号被锁定影响使用。

支持的同步对象范围

在使用Migration进行同步时，不同类型的链路，支持的同步对象范围不同，详细情况可参考下表。

表3 同步对象范围

类型名称

使用须知

同步对象范围

支持同步所有Kafka消息，其中支持对JSON或CSV格式的消息体进行解析。

注意事项

除了数据源版本、连接账号权限及同步对象范围外，您还需要注意的事项请参见下表。

表4 注意事项

类型名称

使用和操作限制

数据库限制
  • 支持开启Kerberos认证的MRS Kafka。
  • 支持使用SASL_PLAINTEXT的Kafka实例，包括SCRAM-SHA-512及PLAIN认证机制。
  • 仅支持使用默认证书的DMS Kafka开启SASL_SSL认证进行数据同步；其余类型Kafka不支持开启SASL_SSL认证进行同步。
  • 目标数据库中Hudi的库名、表名、字段名仅支持数字、字母和下划线，且字段名必须以字母或下划线开头，建议尽量使用常规字符避免任务失败。

使用限制

通用：

  • 实时同步过程中，不支持IP、端口、账号、密码修改。
  • Hudi表使用Bucket索引的场景下不允许更新分区键，否则可能产生重复数据。
  • Hudi表使用Bucket索引的场景下主键仅保证单分区内唯一。

增量同步阶段：

对于版本低于3.2.0-LTS1.5的MRS集群的Hudi MOR表，使用Migration（Flink）增量同步数据后，不支持直接使用CDM或Spark SQL写入数据，需要先进行Compaction后方可写入。

常见故障排查：

在任务创建、启动、全量同步、增量同步、结束等过程中，如有遇到问题，可先参考常见问题章节进行排查。

其他限制

-

操作步骤

本小节以Kafka到MRS Hudi的实时同步为示例，介绍如何配置Migration实时集成作业。配置作业前请务必阅读使用前自检概览， 确认已做好所有准备工作。

  1. 参见新建实时集成作业创建一个实时集成作业并进入作业配置界面。
  2. 选择数据连接类型：源端选Kafka，目的端选Hudi。

    图2 选择数据连接类型

  3. 选择集成作业类型：同步类型默认为实时，同步场景包含单表场景。

    图3 选择集成作业类型

    同步场景相关介绍请参见同步场景。

  4. 配置网络资源：选择已创建的DMS Kafka、MRS Hudi数据连接和已配置好网络连接的migration资源组。

    图4 选择数据连接及migration资源组

    无可选数据连接时，可单击“新建”跳转至管理中心数据连接界面，单击“创建数据连接”创建数据连接，详情请参见配置DataArts Studio数据连接参数进行配置。

    无可选migration资源组时，可单击“新建”跳转至购买migration资源组页面创建migration资源组配置，详情请参见购买创建数据集成资源组增量包进行配置。

  5. 检测网络连通性：数据连接和migration资源组配置完成后需要测试整个迁移任务的网络连通性，可通过以下方式进行数据源和migration资源组之间的连通性测试。

    • 单击展开“源端配置”触发连通性测试，会对整个迁移任务的连通性做校验。
    • 单击源端和目的端数据源和migration资源组中的“测试”按钮进行检测。

      网络连通性检测异常可先参考数据源和资源组网络不通如何排查？章节进行排查。

  6. 配置源端参数。

    • 主题
      输入一个需要迁移的Kafka Topic。
      图5 输入Kafka Topic
    • 数据格式
      源端Kafka Topic中消息内容的格式，Migration当前支持对如下三种消息进行处理：
      • JSON格式：支持对消息内容以JSON的层级格式进行解析。
      • CSV格式：支持对消息内容以CSV格式指定分隔符进行解析。
      • TEXT格式：将整条消息内容作为文本直接同步。
      • DEBEZIUM_JSON：支持解析Debezium格式的CDC JSON，同步数据到目的端对应的库表中。
      • CANAL_JSON：支持解析Canal格式的CDC JSON，同步数据到目的端对应的库表中。
    • 消费组ID

      消费者是从Topic订阅消息的一方，消费组是由一个或多个消费者组成的。Migration支持指定本次消费动作所属的Kafka消费组。

    • Kafka源端属性配置

      支持设置Kafka的配置项，需要增加 properties. 前缀，作业将自动移除前缀并传入底层Kafka客户端，具体参数可参考Kafka官方文档中的配置说明。

    • 上游数据源类型

      仅当数据格式选择DEBEZIUM_JSON或CANAL_JSON时需要配置，当前仅支持选择TiDB，标志CDC JSON为TiDB产生的。

    • 库表名称列表

      仅当数据格式选择DEBEZIUM_JSON或CANAL_JSON时需要配置，标志Kafka消息来源的数据库表，任务会解析CDC JSON并且只同步已配置库表的数据。

  7. 配置目的端参数。

    图6 Hudi目的端配置项
    • 目标表的基本配置

      指定目的端Hudi表所属的库。

    • 目标表名称

      指定目的端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周期任务？。

  8. 刷新源表和目标表映射，检查映射关系是否正确，同时可根据需求修改表属性、添加附加字段，并通过“自动建表”能力在目的端Hudi数据库中建出相应的表。

    源端为CDC JSON（DEBEZIUM_JSON和CANAL_JSON）数据时暂不支持自动建表，用户需手动创建Hudi表。
    图7 源表与目标表映射
    • 同步主键

      Hudi表必须设置“同步主键”，在源端为非主键表时，必须在字段映射阶段手动勾选主键。

    • 分区字段

      目的端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官方文档。

  9. 配置任务属性。

    表7 任务配置参数说明

    参数

    说明

    默认值

    执行内存

    作业执行分配内存，跟随处理器核数变化而自动变化。

    8GB

    处理器核数

    范围：2-32。

    每增加1处理核数，则自动增加4G执行内存和1并发数。

    2

    并发数

    作业执行支持并发数。该参数无需配置，跟随处理器核数变化而自动变化。

    1

    自动重试

    作业失败时是否开启自动重试。

    否

    最大重试次数

    “自动重试”为是时显示该参数。

    1

    重试间隔时间

    “自动重试”为是时显示该参数。

    120秒

    是否写入脏数据

    选择是否记录脏数据，默认不记录脏数据，当脏数据过多时，会影响同步任务的整体同步速度。

    • 否：默认为否，不记录脏数据。

      表示不允许脏数据存在。如果同步过程中产生脏数据，任务将失败退出。

    • 是：允许脏数据，即任务产生脏数据时不影响任务执行。
      允许脏数据并设置其阈值时：
      • 若产生的脏数据在阈值范围内，同步任务将忽略脏数据（即不会写入目标端），并正常执行。
      • 若产生的脏数据超出阈值范围，同步任务将失败退出。
        说明：

        脏数据认定标准：脏数据是对业务没有意义，格式非法或者同步过程中出现问题的数据；单条数据写入目标数据源过程中发生了异常，则此条数据为脏数据。 因此只要是写入失败的数据均被归类于脏数据。

        例如，源端是VARCHAR类型的数据写到INT类型的目标列中，则会因为转换不合理导致脏数据不会成功写入目的端。用户可以在同步任务配置时，配置同步过程中是否写入脏数据，配置脏数据条数（单个分片的最大错误记录数）保证任务运行，即当脏数据超过指定条数时，任务失败退出。

    否

    脏数据策略

    “是否写入脏数据”为是时显示该参数，当前支持以下策略：

    • 不归档：不对脏数据进行存储，仅记录到任务日志中。
    • 归档到OBS：将脏数据存储到OBS中，并打印到任务日志中。

    不归档

    脏数据写入连接

    “脏数据策略”选择归档到OBS时显示该参数。

    脏数据要写入的连接，目前只支持写入到OBS连接。

    -

    脏数据目录

    脏数据写入的OBS目录。

    -

    脏数据阈值

    是否写入脏数据为是时显示该参数。

    用户根据实际设置脏数据阈值。

    说明：
    • 脏数据阈值仅针对每个并发生效。比如阈值为100，并发为3，则该作业可容忍的脏数据条数最多为300。
    • 输入-1表示不限制脏数据条数。

    100

    添加自定义属性

    支持通过自定义属性修改部分作业参数及开启部分高级功能，详情可参见任务性能调优章节。

    -

  10. 提交并运行任务。

    作业配置完毕后，单击作业开发页面左上角“提交”，完成作业提交。

    图8 提交作业

    提交成功后，单击作业开发页面“启动”按钮，在弹出的启动配置对话框按照实际情况配置同步位点参数，单击“确定”启动作业。

    图9 启动配置
    表8 启动配置参数

    参数

    说明

    偏移量参数
    • 最早：从Kafka Topic最早偏移量开始消费数据。
    • 最新：从Kafka Topic最新偏移量开始消费数据。
    • 起止时间：根据时间获取Kafka Topic对应的偏移量，并从该偏移量开始消费数据。

    起始时间

    起止时间需要设置该参数，指示同步起始的时间位点。

    说明：

    配置的位点时间早于Kafka消息最早偏移量时，默认会从最早偏移量开始消费。

  11. 监控作业。

    通过单击作业开发页面导航栏的“前往监控”按钮，可前往作业监控页面查看运行情况、监控日志等信息，并配置对应的告警规则，详情请参见实时集成任务运维。

    图10 前往监控

性能调优

若链路同步速度过慢，可参考参见任务性能调优章节中对应链路文档进行排查及处理。

相关文档