Kafka单表离线同步至MaxCompute

一、创建离线同步节点

1. 进入DataWorks工作空间列表页，在顶部切换至目标地域，找到已创建的工作空间，单击操作列的快速进入 > Data Studio，进入Data Studio。

2. 在左侧导航栏单击，然后单击项目目录右侧的，选择新建节点 > 数据集成 > 离线同步，自定义离线同步任务节点名称后，单击确认。

二、网络与资源配置

1. 在网络与资源配置步骤，选择同步任务所使用的数据来源、资源组、数据去向。您可以为该任务分配任务资源占用CU数。

   • 数据来源选择已添加的Kafka数据源。

   • 数据去向选择已添加的MaxCompute数据源。

   • 资源组选择已与Kafka和MaxCompute数据源连通的资源组。您可以为该任务分配任务资源占用CU数。

2. 在数据来源和数据去向卡片中单击测试连通性。

   

3. 确保数据来源与数据去向均连通成功后，单击下一步。

三、配置数据来源与去向

四、配置字段映射

选择数据来源和数据去向后，需要指定读取端和写入端列的映射关系。您可以选择同名映射、同行映射、取消映射或手动编辑映射关系。

• Kafka侧字段中默认的6个字段。

  字段名	含义
  __key__	Kafka记录的Key。
  __value__	Kafka记录的Value。
  __partition__	Kafka记录所在分区号，分区号为从0开始的整数。
  __headers__	Kafka记录的dHeaders。
  __offset__	Kafka记录在所在分区的偏移量，偏移量为从0开始的整数。
  __timestamp__	Kafka记录的13位整数毫秒时间戳。

• Kafka侧字段可自定义配置JSON解析，可以通过.（获取子字段）和[]（获取数组元素）两种语法，获取Kafka记录JSON格式的value字段内容。

  重要

  如果JSON字段名中带有"."字符，由于会引发字段定义语法歧义，无法通过字段定义获取字段值。

  Kafka某条JSON格式的记录value的数据示例如下。

  {
        "a": {
        "a1": "hello"
        },
        "b": "world",
        "c":[
              "xxxxxxx",
              "yyyyyyy"
              ],
        "d":[
              {
                    "AA":"this",
                    "BB":"is_data"
              },
              {
                    "AA":"that",
                    "BB":"is_also_data"
              }
          ],
       "a.b": "unreachable"
  }

  • 如果同步a1的数据“hello”，Kafka侧字段增加a.a1。

  • 如果同步b的数据“world”，Kafka侧字段增加b。

  • 如果同步c的数据“yyyyyyy”，Kafka侧字段增加c[1]。

  • 如果同步AA的数据“this”，Kafka侧字段增加d[0].AA。

  • Kafka侧字段定义增加a.b无法同步数据"unreachable"。

• 允许源头表字段或目标表字段存在不参与映射的字段，源头表不参与映射的字段同步实例不会读取，目标端不参与映射的字段将写入NULL。

• 不允许一个源头表字段映射到多个目标表字段，也不允许一个目标表字段映射到多个目标表字段

五、配置通道控制

离线同步任务支持设置任务期望最大并发数、脏数据策略等。本教程脏数据策略配置为不容忍脏数据，其他配置保持默认。更多信息，请参见通过向导模式配置离线同步任务。

六、调试配置并运行

1. 单击离线同步节点编辑页面右侧的调试配置，设置调试运行使用的资源组和脚本参数，然后单击顶部工具栏的运行，测试同步链路是否成功运行。

2. 您可以在左侧导航栏单击，然后单击个人目录右侧的，创建一个后缀为.sql的文件，执行如下SQL查询数据去向表中的数据是否符合预期。

   说明

   • 此方式查询需要将目标端MaxCompute绑定为DataWorks的计算资源。

   • 您需要在.sql文件编辑页面右侧单击调试配置，指定数据源类型、计算资源、资源组后，再单击顶部工具栏的运行。

   SELECT * FROM <MaxCompute侧目标表名> WHERE pt=<指定分区> LIMIT 20;

七、调度配置与发布

单击离线同步任务右侧的调度配置，设置周期运行所需的调度配置参数后，单击顶部工具栏的发布，进入发布面板，根据页面提示完成发布。

在上述配置数据来源与数据去向时，使用了三个调度参数：${startTime}、${endTime}和${partition}，在此处调度配置中需根据实际同步需求指定这三个调度参数的替换策略，以下为几个典型场景的配置示例。

根据希望的调度间隔，设置调度周期。