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

前提条件

创建数据同步任务前,需检查已完成以下准备操作。
  • 资源规划与配置
  • 配置数据源(来源为MySQL)
  • 配置数据源(来源为Oracle)
  • 配置数据源(来源为PolarDB)
  • 添加数据源
  • 独享集成资源组离线同步插件Datax版本必须大于等于20210726203000,实时同步插件Streamx版本必须大于等于202107121400,否则运行增量和全量同步Kafka数据时可能失败或者存在数据格式错误。
    离线同步插件版本:在运维中心>离线同步任务日志中搜索Detail log url跳转到离线同步详情日志页,然后搜索DataX( ..... ),From Alibaba!格式文本,例如,DataX (20210709_keyindex-20210709144909), From Alibaba !”,截图中框起来的内容即离线同步插件Datax的版本。离线插件版本查看方法离线版本查看方法2离线版本查看方法3
    实时同步插件版本:在运维中心>实时同步任务日志中搜索StreamX( ..... ),From Alibaba!格式文本,例如,StreamX (202107290000_20210729121213), From Alibaba !,截图中内容即实时同步插件StreamX的版本。实时同步插件版本查看方法

创建同步解决方案任务

  1. 登录并进入数据集成页面,单击同步解决方案 > 任务列表,进入同步解决方案页面。
    操作详情可参见选择同步解决方案
  2. 解决方案任务列表页面,单击右上方的新建任务
  3. 选择同步方案。
    1. 选择数据来源与去向。
      数据来源数据去向区域,选择数据源类型。
      说明 一键实时同步至Kafka方案仅支持选择MySQL、Oracle和PolarDB类型的数据源。
      同步方案
    2. 选择可用的同步方案区域,单击一键实时同步至Kafka
    3. 单击下一步
  4. 配置同步网络链接。
    1. 选择数据来源数据去向数据源名称。如果下拉列表中没有数据源,单击新建数据源进行创建,详情请参见:配置MySQL数据源配置Oracle数据源配置PolarDB数据源配置同步网路链接
    2. 我的资源组区域选择独享数据集成资源组。如果列表中没有资源组,单击新建独享数据集成资源组,选择相应规格资源数量计费周期,单击确认购买跳转到支付页面完成支付,请参见:资源规划与配置
      说明

      地域默认是该工作空间所在区域。

      购买后独享集成资源组默认与该工作空间绑定。

    3. 单击连通性测试验证独享数据集成资源组和数据源的网络连通性。详情请参见:选择网络连通方案。如果网络不通,请根据弹出的网络连通性诊断工具逐步诊断。
    4. 单击下一步
  5. 选择来源数据源并配置同步规则。
    1. 完成方案名称等基本信息配置。
      基本配置区域,配置各项参数。基本配置
      参数 描述
      方案名称 同步解决方案的名称,最多支持50个字符。
      描述 对当前方案进行简单描述,最多支持50个字符。
      目标任务存放位置 默认创建一个新的业务流程,所有任务均以clone_database_源端数据源名称+to+目标数据源名称的命名方式存放至数据集成目录下。

      您也可以取消自动建立工作流程,在选择位置下拉列表中指定存放目标任务的路径。

    2. 数据来源区域,选择数据来源的编码格式。
    3. 选择同步的源表区域,选中需要同步的源端库表,单击图标图标,将其移动至已选库表
      原表该区域会为您展示所选数据源下所有的表,您可以选择整库全表或部分表进行同步。
    4. 设置表名到Topic的映射规则区域,单击添加规则,选择相应的规则进行添加
      同步规则包括源表名和目标Topic转换规则目标Topic规则
      • 源表名和目标Topic转换规则:转换表名为目标表名,进行字符串替换。
      • 目标Topic规则:支持对转换后的表名添加前缀和后缀。
    5. 单击下一步
  6. 配置目标Topic格式。
    1. 目标Kafka数据源默认为已配置好的数据源。
    2. 单击刷新源表和Kafka Topic映射,创建需要同步的源表和目标Kafka Topic的映射关系。
    3. 查看任务的执行进度和表来源。查看任务进度
      序号 描述
      显示映射关系的创建进度。
      说明 如果同步的表数量较多,会导致执行进度较慢,请耐心等待。
      • 勾选支持源表无主键同步后,源表没有主键,也可以向下游同步,但是同步数据时不会进行去重。
      • 勾选发送心跳记录后,实时同步任务将每隔5秒往Kafka中写入一条带有当前时间信息的记录。这样即使源端没有读取到新数据,Kafka中最新数据的时间信息也会持续更新,您可以根据Kafka中读取到的最新数据的时间判断实时同步的进度。
      • 如果来源库有主键,同步数据时会直接使用该主键进行去重。
      • 如果勾选支持源表无主键同步,那么源表没有主键,您需要单击编辑图标,自定义主键,即使用其他非主键的一个或几个字段的联合,代替主键进行同步数据时进行去重判断。
      包括使用已有Topic自动建Topic

      选择的Topic建立方式,取值如下:

      • Topic建立方式选择使用已有Topic时,您可以在Kafka Topic列的下拉列表中选择需要使用的Topic名称。
      • Topic建立方式选择自动建Topic时,显示自动创建的Kafka Topic名称。您可以单击Topic名称,查看和修改建Topic名称和注释。
      批量编辑目标Topic附加字段表单中给目标Kafka Topic增加字段。也可以单击操作列的编辑附加字段进行单表附加字段的设置。
      说明 批量修改仅针对Topic建立方式选择自动建Topic的Topic生效。
    4. 单击下一步
  7. 运行资源设置。
    运行资源设置页面,配置各项参数。运行资源设置
    • 离线全量同步
      参数 描述
      离线任务名称规则 全量同步时的离线任务名称。创建解决方案后,会先生成一个离线任务用于同步全量数据,再生成实时任务实时同步增量数据。
      全量离线任务资源组

      运行全量离线任务需要使用的独享数据集成资源组。

      目前解决方案仅支持使用独享数据集成资源组,此处可配置为准备操作中已购买并配置的独享数据集成资源组,详情请参见资源规划与配置
      说明 如果您没有购买独享资源组,可单击+新建独享资源组,创建新的独享资源组。
    • 离线全量调度
      参数 描述
      选择调度资源组

      选择运行任务时使用的调度资源组。

      目前解决方案仅支持使用独享调度资源组,此处可配置为准备操作中已购买并配置的独享调度资源组,详情请参见资源规划与配置
      说明 如果您没有购买独享资源组,可单击+新建独享资源组,创建新的独享资源组。
    • 实时增量同步
      参数 描述
      选择实时任务独享资源组

      选择运行实时任务时需要使用的独享数据集成资源组。

      目前解决方案仅支持使用独享数据集成资源组,此处可配置为准备操作中已购买并配置的独享数据集成资源组,详情请参见资源规划与配置
      说明 如果您没有购买独享资源组,可单击+新建独享资源组,创建新的独享资源组。
    • 通道设置
      参数 描述
      来源端读取支持最大连接数 读取端的最大连接数,即来源端数据库的JDBC连接数。请根据数据库资源的实际情况合理配置。默认为20
  8. 单击完成配置,完成整库实时同步任务的创建。

执行数据同步解决方案任务

解决方案任务列表页面,单击相应任务后的提交执行,运行创建的数据同步解决方案任务。

查看运行状态及结果

  • 解决方案任务列表页面,单击已运行任务后的执行详情,查看当前解决方案数据同步过程中各子任务节点的运行详情。执行详情
  • 单击子任务节点后的执行详情,可在弹窗中单击任务链接进入子节点的数据开发页面。

管理数据同步解决方案任务

  • 查看或编辑任务。

    解决方案任务列表页面,单击相应任务后的更多 > 查看配置更多 > 修改配置,可查看或编辑任务的配置信息。

  • 删除任务。
    单击相应任务后的更多 > 删除。在删除对话框中,单击确定
    说明 仅删除当前任务的配置记录,已经生成的表和任务不受影响。
  • 修改任务优先级。
    单击相应任务后的更多 > 修改优先级。在修改优先级对话框中,输入需要配置的优先级数值,单击确定。优先级取值范围为1~8,数值越大优先级越高。
    说明 优先级相同的任务,按照提交时间的先后顺序执行。

写入Kafka消息格式定义

完成配置实时同步任务的操作后,执行同步任务会将源端数据库读取的数据,以JSON格式写入到Kafka topic中。除了会将设置的源端表中已有数据全部写入Kafka对应Topic中,还会启动实时同步将增量数据持续写入Kafka对应Topic中,同时源端表增量DDL变更信息也会以json格式写入Kafka对应Topic中。您可以通过附录:消息格式获取写入Kafka的消息的状态及变更等信息。
说明 通过离线同步任务写入Kafka的数据json结构中的payload.sequenceId、payload.timestamp.eventTImepayload.timestamp.checkpointTime字段均设置为-1