Apache Kafka同步到 MRS Kafka作业配置

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

表1 支持的数据库版本
源端数据库	目的端数据库
Kafka集群（2.7、3.x版本）	Kafka集群（2.7、3.x版本）

数据库账号权限要求

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

表2 数据库账号权限
类型名称	权限要求
源数据库连接账号	-
目标数据库连接账号	开启密文接入场景下，所配置用户需要有发布和订阅Topic的权限，其余场景无特殊权限要求。说明：若为MRS Kafka，则所使用的MRS用户必须属于kafka/kafkaadmin/kafkasuperuser用户组，才能拥有Kafka对应Topic的读写权限。图1 对应权限

建议创建单独用于Migration任务连接的数据库账号，避免因为数据库账号密码修改，导致的任务连接失败。
连接源和目标数据库的账号密码修改后，请同步修改管理中心对应的连接信息，避免任务连接失败后自动重试，导致数据库账号被锁定影响使用。

支持的同步对象范围

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

表3 同步对象范围
类型名称	使用须知
同步对象范围	支持完整同步Kafka Topic所有消息内容，但不支持对Kafka Topic消息进行解析重组后同步。

注意事项

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

表4 注意事项
类型名称	使用和操作限制
数据库限制	支持开启/未开启Kerberos认证的MRS集群Kafka实例。仅支持使用默认证书的DMS Kafka开启SASL_SSL认证进行数据同步；其余类型Kafka不支持开启SASL_SSL认证进行同步。
使用限制	通用：实时同步过程中，不支持IP、端口、账号、密码修改。增量同步阶段：整库场景下需要根据同步的Topic分区数对应增加作业并发数，否则可能导致任务内存溢出。常见故障排查：在任务创建、启动、全量同步、增量同步、结束等过程中，如有遇到问题，可先参考常见问题章节进行排查。
其他限制	-

操作步骤

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

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

图2 选择数据连接类型
选择集成作业类型：同步类型默认为实时，同步场景仅支持整库场景。

图3 选择集成作业类型

同步场景相关介绍请参见同步场景。
配置网络资源：选择已创建的Kafka数据连接和已配置好网络连接的migration资源组。

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

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

无可选migration资源组时，可单击“新建”跳转至购买migration资源组页面创建migration资源组配置，详情请参见购买创建数据集成资源组增量包进行配置。
检测网络连通性：数据连接和migration资源组配置完成后需要测试整个迁移任务的网络连通性，可通过以下方式进行数据源和migration资源组之间的连通性测试。
- 单击展开“源端配置”触发连通性测试，会对整个迁移任务的连通性做校验。
- 单击源端和目的端数据源和migration资源组中的“测试”按钮进行检测。
  
  网络连通性检测异常可先参考数据源和资源组网络不通如何排查？章节进行排查。
配置源端参数。
- 选择需要同步的Kafka Topic。
  图5 选择需要同步的Kafka Topic
- 消费组ID
  消费者是从Topic订阅消息的一方，消费组是由一个或多个消费者组成的。Migration支持指定本次消费动作所属的Kafka消费组。
- Kafka源端属性配置
   支持设置Kafka的配置项，需要增加 properties. 前缀，作业将自动移除前缀并传入底层Kafka客户端，具体参数可参考Kafka官方文档中的配置说明。
配置目的端参数。

图6 Kafka目的端配置项
- 目标Topic名称规则。
  配置源端MySQL库表与目的端Kafka Topic的映射规则。可指定为固定的一个Topic，也可使用内置变量做映射，将不同源表数据同步到不同的Topic中。
  
  可以使用的内置变量有：源Topic名：#{source_Topic_name}
- 同步kafka partition策略
  支持以下三种投递策略将源端的数据按规则同步到Kafka Topic的特定Partition：
  - 全部投递到Partition 0。
  - 按源端分区投递到对应的Partition：源端消息在第n个分区，则投递到目的端的第n个分区，该策略可以保证消息顺序。
  - 按轮询模式投递到不同的Partition：采用Kafka粘性分区策略均匀的投递到目的端主题的所有分区，该策略无法保证消息顺序。
- 新建Topic的Partition数量
   设定目的端Kafka无对应Topic时，Migration自动建Topic的分区数量，默认为3。
- Kafka目标端属性配置
   支持设置Kafka的配置项，需要增加 properties. 前缀，作业将自动移除前缀并传入底层Kafka客户端，具体参数可参考Kafka官方文档中的配置说明。
刷新源表和目标表映射，检查源端Topic和目的端Topic映射关系是否正确，支持用户根据实际需求修改映射后的目的端Topic名称，可以配置为一对一、多对一的映射关系。

图7 源表与目标表映射

配置任务属性。

表5 任务配置参数说明
参数	说明	默认值
执行内存	作业执行分配内存，跟随处理器核数变化而自动变化。	8GB
处理器核数	范围：2-32。每增加1处理核数，则自动增加4G执行内存和1并发数。	2
并发数	作业执行支持并发数。该参数无需配置，跟随处理器核数变化而自动变化。	1
自动重试	作业失败时是否开启自动重试。	否
最大重试次数	“自动重试”为是时显示该参数。	1
重试间隔时间	“自动重试”为是时显示该参数。	120秒
添加自定义属性	支持通过自定义属性修改部分作业参数及开启部分高级功能，详情可参见任务性能调优章节。	-

提交并运行任务。

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

图8 提交作业

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

图9 启动配置

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