DMS Kafka同步到OBS

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

数据库账号权限要求

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

支持的同步对象范围

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

注意事项

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

操作步骤

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

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

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

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

4. 配置网络资源：选择已创建的DMS Kafka、OBS数据连接和已配置好网络连接的资源组。
   图3 选择数据连接及资源组
   

5. 检测网络连通性：数据连接和资源组配置完成后需要测试整个迁移任务的网络连通性，可通过以下方式进行数据源和资源组之间的连通性测试。
   • 单击展开“源端配置”触发连通性测试，会对整个迁移任务的连通性做校验。
   • 单击源端和目的端数据源和资源组中的“测试”按钮进行检测。
     

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

6. 配置源端参数。
   • 选择需要同步的Kafka Topic各同步场景下选择需要同步库表的方式请参考下表。

     表5 选择需要同步的库表

     同步场景	配置方式
     单表	输入一个需要迁移的Kafka Topic。 图4 输入Kafka Topic
     整库	选择需要迁移的Kafka Topic。 图5 选择Kafka Topic

   • 数据格式
     源端Kafka Topic中消息内容的格式，Migration当前支持对如下三种消息进行处理：

     • JSON格式：支持对消息内容以JSON的层级格式进行解析。
     • CSV格式：支持对消息内容以CSV格式指定分隔符进行解析。
     • TEXT格式：将整条消息内容作为文本直接同步。
   • 消费组ID

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

   • Kafka源端属性配置

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

7. 配置目的端参数。
   图6 目的端OBS配置
   
   • 文件存储格式

     写入OBS的文件格式，当前支持Parquet、SequenceFile和TextFile。

   • 文件压缩方式
     指定写入OBS文件的压缩方式，默认不进行压缩，支持以下列表：

     • Parquet格式：UNCOMPRESSED、SNAPPY。
     • SequenceFile格式：UNCOMPRESSED、SNAPPY、GZIP、LZ4、BZIP2。
     • TextFile格式：UNCOMPRESSED。
   • OBS存储路径

     指定OBS文件存储的路径，支持填写#{source_Topic_name}内置变量，可将源端不同的Topic的数据写入不同的路径下，例如：obs://bucket/dir/test.db/prefix_#{source_Topic_name}_suffix/

   • 全局高级配置
     支持通过参数配置部分高级功能，详情请参考下表。

     表6 OBS高级配置一览表

     参数名	参数类型	默认值	单位	参数说明
     auto-compaction	boolean	false	-	文件自动合并开关。数据会先被写入临时文件，当checkpoint完成后，该配置控制检查点内产生的临时文件是否被合并。开启该配置部分场景下可减少小文件数量，但会较大降低同步速率。

8. 刷新源表和目标表映射，单击“目标字段编辑”检查要写入目的端的字段情况，并根据实际情况选择配置分区字段。
   图7 源表与目标表映射
   
   • 分区字段

     支持配置分区字段，将在写入OBS时自动生成对应分区目录，目录名为“分区字段=分区值”。同时，字段选择顺序影响分区的层级，例如选择par1、par2作为分区字段，那么par1为一级分区，par2为二级分区，最多支持五级分区。

   • 目标字段编辑
     Migration会根据选择的源端消息格式自动解析源端消息，生成对应的字段信息，用户可在此基础上进行编辑，自定义字段名、选择字段类型、填写字段值。

     • 字段名称：目的端OBS文件中写入字段的名称。字段名称至少包含一个字母，允许下划线、中划线，不支持纯数字。
     • 字段类型：目的端OBS文件中写入字段的类型。当前支持STRING、BOOLEAN、INTEGER、LONG、FLOAT、DOUBLE、SHORT、DECIMAL、DATE、TIMESTAMP类型。
     • 字段值：目的端OBS文件中写入字段的取值来源。

       表7 目标字段取值方式

       类型	字段取值
       手动赋值	任意字符。
       内置变量	Kafka的元数据，包括__key__、__value__、__Topic__、__partition__、__offset__、__timestamp__共6个字段。
       源表字段	从源端Kafka Topic消息解析出的任一字段。 说明： 如果源端Kafka消息为嵌套JSON的形式，本链路支持解析不同层级的字段值（包含数组，数组下标索引从1开始）。 例如，JSON的内容为： { "col1": "1", "col2": "2", "level1": { "level2": [ { "level3": "test" } ] } } 则可以通过level1.level2[1].level3取到数据”test”作为目标端某一个字段的值。
       udf方法	支持填写Flink的内置函数用于数据转换，例如： CONCAT(CAST(NOW() as STRING), `col_name`) DATE_FORMAT(NOW(), 'yy') 注意，其中的字段名要用反引号包围起来。Flink完整内置函数可参考Flink官方文档。

9. 配置任务属性。

   表8 任务配置参数说明

   参数	说明	默认值
   执行内存	作业执行分配内存，跟随处理器核数变化而自动变化。	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 启动配置
    

    表9 启动配置参数

    参数	说明
    偏移量参数	最早：从Kafka Topic最早偏移量开始消费数据。 最新：从Kafka Topic最新偏移量开始消费数据。 起止时间：根据时间获取Kafka Topic对应的偏移量，并从该偏移量开始消费数据。
    时间	起止时间需要设置该参数，指示同步起始的时间位点。 说明： 配置的位点时间早于Kafka消息最早偏移量时，默认会从最早偏移量开始消费。

11. 监控作业。

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

    图10 前往监控
    

性能调优

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