完成数据源、网络、资源的准备配置后,您可创建实时同步节点,同步数据至Kafka。本文为您介绍如何创建数据实时同步任务,并在创建完成后查看任务运行情况。

前提条件

创建实时数据同步任务前,需检查已完成以下准备操作。

使用限制

  • 实时数据同步任务仅支持使用独享数据集成资源组。
  • 实时同步节点目前仅支持同步MySQL数据源至Kafka。

创建实时同步任务

  1. 登录DataWorks控制台
  2. 在左侧导航栏,单击工作空间列表
  3. 选择工作空间所在地域后,单击相应工作空间后的进入数据开发
  4. 创建业务流程
    如果您已有 业务流程,则可以忽略该步骤。
    1. 鼠标悬停至新建图标,选择业务流程
    2. 新建业务流程对话框,输入业务名称
    3. 单击新建
  5. 创建实时同步节点。
    1. 鼠标悬停至新建图标,选择数据集成 > 实时同步
      您也可以找到目标业务流程,右键单击 数据集成,选择 新建 > 实时同步
    2. 新建节点对话框中,配置各项参数。新建实时同步节点
      参数 描述
      节点类型 默认为实时同步
      同步方式 选择整库实时同步至Kafka,用于迁移目标数据库下的部分或所有表至Kafka中。
      节点名称 节点名称必须是大小写字母、中文、数字、下划线(_)以及英文句号(.),且不能超过128个字符。
      目标文件夹 存放实时同步节点的目录。
    3. 单击提交,进入实时同步任务编辑页面。
  6. 选择资源组。
    1. 在实时同步任务编辑页面的右侧导航栏,单击基本配置
    2. 资源组下拉框,选择需要使用的资源组。
      说明

      实时数据同步任务仅支持使用独享数据集成资源组。

      如果您没有可用的独享数据集成资源组,请单击 新建独享资源组创建。详情请参见 独享数据集成资源组
  7. 选择来源数据源并配置同步规则。
    1. 数据来源区域,选择类型数据源编码格式。
      说明 仅支持选择MySQL、Oracle和PolarDB数据源。
    2. 选择同步的源表区域,选中需要同步的源表,单击图标图标,将其移动至已选源表选择同步原表
      该区域会为您展示所选数据源下所有的表,您可以选择整库全表或部分表进行同步。
    3. 择同步的源表区域搜索框下方,勾选使用上述输入文字作为正则表达式自动加表复选框,仅MySQL支持通过正则表达自动加表。
      勾选后可以在源端库表的 库过滤搜索框和 表过滤搜索框中填写正则表达式,实时同步运行过程中会自动识别源端数据库binlog中数据库名和表名符合正则表达式的变更,识别到这类变更后按照下面设置的 表名到Topic的映射规则计算源端变更应该写入的 kafka topic名称,如果kafka topic尚未建立,则会自动创建对应topic后将变更写入。 正则表达式

      库名及表名支持的正则规则如下所示。

      正则表达式 描述
      . 匹配除换行符以外的所有字符
      x? 匹配 0 次或一次 x 字符串
      x* 匹配 0 次或多次 x 字符串,但匹配可能的最少次数
      x+ 匹配 1 次或多次 x 字符串,但匹配可能的最少次数
      .* 匹配 0 次或多次的任何字符
      .+ 匹配 1 次或多次的任何字符
      {m} 匹配刚好是 m 个 的指定字符串
      {m,n} 匹配在 m个 以上 n个 以下 的指定字符串
      {m,} 匹配 m个 以上 的指定字符串
      [] 匹配符合 [] 内的字符
      [^] 匹配不符合 [] 内的字符
      \d 匹配一个数字的字符,和 [0-9] 语法一样
      \d+ 匹配多个数字字符串,和 [0-9]+ 语法一样
      \D 非数字,其他同 \d
      \D+ 非数字,其他同 \d+
      \w 英文字母或数字的字符串,和 [a-zA-Z0-9_] 语法一样
      \w+ 和 [a-zA-Z0-9_]+ 语法一样
      \W 非英文字母或数字的字符串,和 [^a-zA-Z0-9_] 语法一样
      \W+ 和 [^a-zA-Z0-9_]+ 语法一样
      \s 空格,和 [\n\t\r\f] 语法一样
      \s+ 和 [\n\t\r\f]+ 一样
      \S 非空格,和 [^\n\t\r\f] 语法一样
      \S+ 和 [^\n\t\r\f]+ 语法一样
      \b 匹配以英文字母,数字为边界的字符串
      \B 匹配不以英文字母,数值为边界的字符串
      (a|b|c) 匹配符合a字符 或是b字符 或是c字符 的字符串,注意或表达式必须用括号包围,否则将导致运行时错误
      \ 对"$^[].?+|\*{}"具有特殊含义的正则规则字符转义
    4. 设置表名到Topic的映射规则区域,单击添加规则,选择相应的规则进行添加。
      映射规则包括 源表名和目标Topic转换规则目标Topic规则
      • 源表名和目标Topic转换规则:转换表名为目标Topic,进行字符串替换。
      • 目标Topic规则:支持对转换后的Topic添加前缀和后缀。
    5. 单击下一步
  8. 选择目标数据源并配置目标Topic格式。
    1. 设置目标Topic页面,选择目标Kafka数据源
      • 如果您需要同步无主键的来源表,则可以勾选支持源表无主键同步
      • 如果您希望消费Kafka中数据时能够准确判断同步任务进度,则可以勾选发送心跳记录,同步任务时将定时产生op字段作为MHEARTBEAT的同步任务心跳记录写入Kafka的topic中,心跳记录具体格式请参考附录:消息格式
    2. 可选:目标表新增字段。
      如果您希望为目标表中的所有同步表新增统一的字段,则可以在 目标表附加常量字段区域,单击 新增字段添加。
    3. 单击刷新源表和Kafka Topic映射,创建需要同步的源表和目标Kafka Topic的映射关系。
    4. 查看任务的执行进度和表来源。查看任务进度和表来源
      序号 描述
      显示映射关系的创建进度。
      说明 如果同步的表数量较多,会导致执行进度较慢,请耐心等待。
      • 如果来源库有主键,则同步数据时会直接使用该主键进行去重。
      • 如果来源库没有主键:
        • 当在设置目标Topic页面勾选了支持源表无主键同步,则无主键的表可以正常同步。您可以选择单击编辑图标,自定义主键,即使用其他非主键的一个或几个字段的联合,代替主键进行同步数据时进行去重判断。
        • 当在设置目标Topic页面未勾选支持源表无主键同步,则无主键的表同步时会出现异常,您需要在同步任务中删除无主键的表,才能继续执行同步任务。
      选择的Topic建立方式不同,此处显示的Kafka Topic不同:
      • Topic建立方式选择自动建Topic时,显示自动创建的Topic名称。您可以单击Topic名称,编辑Topic信息。
      • Topic建立方式选择使用已有Topic时,请在下拉列表中选择需要的Topic。
    5. 单击下一步
      如果您前一步中目标数据源使用的 Topic建立方式自动建Topic,则需要在弹出的 自动建表对话框,单击 开始建表,批量创建目标Kafka Topic。
  9. 运行资源设置。
    1. 运行资源设置页面,配置各项参数。
      参数 描述
      来源端读取支持最大连接数 读取端的最大连接数,即来源端数据库的JDBC连接数。请根据数据库资源的实际情况合理配置。默认为15
      目标端写入并发数 数据同步任务内,可以从来源表并行读取或写入数据至目标端的最大线程数。最大并发数为32。请根据您的资源组大小和目标端实际规模合理设置。
    2. 单击完成配置

提交并发布实时同步任务

提交并发布节点任务。
  1. 单击工具栏中的保存图标,保存节点。
  2. 单击工具栏中的提交图标,提交节点任务。
  3. 提交新版本对话框中,输入变更描述
  4. 单击确认
如果您使用的是标准模式的工作空间,任务提交成功后,需要将任务发布至生产环境进行发布。请单击右上方的 任务发布。具体操作请参见 发布任务

执行实时同步任务

  1. 进入运维页面。
    提交或发布节点成功后,单击节点编辑页面右上方的 运维中心,进入 实时任务运维 > 实时同步任务页面。
  2. 查看实时同步任务详情。
    实时同步任务页面,单击相应 任务名称,查看运维任务的详细信息。 实时任务运维详情
  3. 执行实时同步任务。
    1. 单击目标实时同步任务操作列的启动
    2. 启动对话框中,配置各项参数。启动实时同步任务
      参数 描述
      是否重置位点 如果选中该参数,请设置下次启动的时间位点。即启动时间点位时区为必选项。
      启动时间点位 选择启动节点任务的日期和时间。
      时区 时区下拉列表中选择执行任务的时区。
      Failover 您可以设置在固定时间内,任务的Failover超过指定次数时,自动结束任务。
      说明 如果您不配置Failover的次数,将根据5分钟Failover超过100次来自动结束任务,避免频繁启动任务占用系统资源。
      脏数据策略
      • 零容忍,不允许:只要同步任务中包含脏数据,则任务自动结束。
      • 不限制:无论同步任务中是否包含脏数据,任务均可正常执行。
      • 有限控制:指定可包含固定数值的脏数据,超出该数值时任务自动结束。
    3. 单击确定

管理实时同步任务

  • 停止运行中的任务。

    单击相应任务后的停止。在停止对话框中,单击停止

  • 下线非运行状态的任务。

    单击相应任务后的下线。在下线对话框中,单击下线

  • 查看任务的报警信息。

    单击相应任务后的报警设置,在报警设置页面查看报警事件报警规则

  • 为任务新增告警。
    1. 选中需要新增告警的任务,单击实时同步任务页面下方的新增告警
    2. 新建规则对话框中,配置各项参数。新建规则
      参数 描述
      名称 新建规则的名称。
      描述 新建规则的描述信息。
      指标 产生报警的指标项:
      • 任务状态
      • 业务延迟
      • Failover
      • 脏数据
      • DDL错误
      阈值 设置WARNINGCRITICAL的阈值,默认值为5分钟。
      报警间隔 设置报警的时间间隔,默认为5分钟发一次报警。
      WARNING 产生相应报警时,可以选择通过邮件短信电话钉钉发送报警通知。
      CRITICAL
      接收人(非钉钉) 选择报警通知的接收人。
    3. 单击确定
  • 批量修改目标任务中指定类型的所有告警。
    1. 选中需要操作告警的任务,单击实时同步任务页面下方的操作告警
    2. 操作告警对话框中,选则需要修改的操作类型告警指标
    3. 单击确定

后续步骤

完成配置实时同步任务的操作后,执行同步任务会将源端数据库读取的数据,以JSON格式写入到Kafka topic中,您可以通过 附录:消息格式获取写入Kafka的消息的状态及变更等信息。