MySQL_to_Doris 整库实时

一键实时整库同步方案支持全增量一体化同步，本实践中，先将 MySQL 源端全量数据通过离线任务同步方式迁移，然后再通过实时同步增量任务，将增量数据采集至目标端 Doris 数据库表中。
您也可以选择单独进行实时增量数据同步，该场景适用于您已经通过其他方式将全量数据同步至 Doris，本文为您介绍如何创建一键实时同步，将全量和增量数据同步至 Doris 任务。

1 前置操作

1. 已开通并创建 DataLeap 项目，创建的全量增量任务均会同步到该项目下。详见新建项目。

2. 已创建合适资源规格的独享数据集成资源组，并将其绑定至创建成功的 DataLeap 项目下。购买操作详见资源组管理，项目绑定操作详见数据集成资源组。

3. 已开通全域数据集成（DataSail）产品。详见服务开通。

4. 已完成数据源来源和目标端的准备，创建源端数据源 MySQL 实例，目标端火山引擎 E-MapReduce（EMR）Doris 集群类型。详见创建 MySQL 实例和 创建 Doris 集群。

2 方案关键步骤概述

1. 目标表创建：将源端表数据写入到目标表中，目标表创建可以是以下两种方式：

   1. 已有表：手动在目标数据库环境中，已创建好源端同名表来接收数据，此时方案步骤执行时，将跳过建表的流程。

   2. 自动建表：在目标数据库环境中，还没有与源端数据表同名的目标表，此时方案步骤执行时，会自动在流程中创建同名目标表。

2. 增量实时任务位点初始化：首次启动时，会自动指定增量流式任务的初始化位点位置。

3. 任务创建：MySQL_to_Doris 的实时整库全增量同步方案，会创建以下两种任务类型：

   1. 一次性全量批式任务创建：全量同步时，一次性全量批式任务将历史全量数据，同步至目标 Doris 表中。

   2. 增量流式任务创建：待全量批式任务执行完成后，增量数据将通过启动流式任务的方式，实时同步至目标 Doris 表中。

   说明

   同步方案产生的一次性全量批式任务个数，与方案中设置的数据来源表个数有关。

3 注意事项

1. 同步解决方案同时支持选择的表数量目前上限为 1000 张，但建议先以 100 张以下表数量来试用。

2. 目前目标库 Doris 库需要提前在集群中先创建好，暂不支持在解决方案中自动创建。

3. 目前表建立方式为选用已有表时，需要保证表 schema 和源表 schema 的名称和数量保持一致。

4. 实时多表目前只支持源表和目标表名完全一致。

5. 解决方案同步数据至 Doris 时，仅支持将数据写入非分区表，暂不支持写入数据至分区表。

6. 支持采集 EMR-3.1.1 及以上 Doris 集群版本的数据，其中 EMR-3.1.1 集群仅支持在解决方案中使用已有表，EMR-3.2.1 支持使用已有表和自动建表功能。

4 数据同步解决方案

4.1 数据源配置

在配置实时整库同步解决方案前，您需在数据源管理界面中，配置来源端 MySQL 和目标端 Doris 相应的数据源。详见 MySQL 数据源配置、Doris 数据源配置。

• 独享数据集成资源组所在的 VPC 需要和 MySQL 实例、Doris 集群所在的 VPC 保持一致，且需要将 VPC 中的 IPv4 CIDR 地址，加入到火山引擎 RDS-MySQL 数据库实例的白名单下，保证资源组与数据源之间的网络互通；

  

• 若资源组和数据源实例不在同一 VPC 环境时，您可通过公网或者通过专线形式进行互通。网络配置详见网络连通解决方案。

4.2 模拟数据准备

源端 MySQL 数据库中，创建以下表结构，并插入 5 条示例数据：

• 创建 MySQL 表：

  CREATE TABLE `mysql_2_doris_cdc`
  (
      `Id`             bigint(20) unsigned NOT NULL AUTO_INCREMENT,
      `Name`           char(10)                     DEFAULT NULL,
      `Address`        varchar(400)                 DEFAULT NULL,
      `Event_Time`     bigint(20)                   DEFAULT NULL,
      `Price`          double                       DEFAULT NULL,
      `Num_Big`        bigint(20)                   DEFAULT NULL,
      `Num_Float`      float(13, 3)                 DEFAULT NULL,
      `Datetime_Info`  datetime                     DEFAULT NULL,
      `Timestamp_Info` timestamp           NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
      `Tinytext_Info`  tinytext,
      PRIMARY KEY (`Id`)
  ) ENGINE = InnoDB
    AUTO_INCREMENT = 456014264
    DEFAULT CHARSET = utf8;
  

• 插入模拟示例数据：

  insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info, Timestamp_Info,
                               Tinytext_Info)
  values (1, 'dataleap', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
          -19505.969, '2023-03-27 14:24:22', '2023-03-27 14:24:22',
          'KKKKKKKKKK');
  
  insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                               Tinytext_Info)
  values (2, '数据开发', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
          -19505.969, '2023-03-27 14:24:22',
          'KKKKKKKKKK');
  
  insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                               Tinytext_Info)
  values (3, '数据集成', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
          -19505.969, '2023-03-27 14:24:22',
          'KKKKKKKKKK');
  
  insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                               Tinytext_Info)
  values (4, '数据运维', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
          -19505.969, '2023-03-27 14:24:22',
          'KKKKKKKKKK');
  
  insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                               Tinytext_Info)
  values (5, '数据监控', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
          -19505.969, '2023-03-27 14:24:22',
          'KKKKKKKKKK');
  

4.3 新建数据实时整库同步

数据源配置操作准备完成后，您可开始进行实时整库同步方案配置：

1. 登录 DataSail 控制台。

2. 在左侧导航栏中选择数据同步方案，进入同步方案配置界面。

3. 单击目录树中项目选择入口，选择已创建的 DataLeap 项目。

   

4. 单击右上角新建数据同步解决方案按钮，下拉选择实时整库同步按钮，进入整库实时同步方案配置界面。按照以下配置，完成方案新建。

4.2.1 基本配置

基本配置参数说明如下表所示。其中名称前带 * 的参数为必填参数，名称前未带 * 的参数为可选参数。

方案基本配置完成后，单击右下角下一步按钮，进行方案的数据来源配置。

4.2.2 数据来源配置

在数据来源配置界面中，完成数据来源设置与库表映射规则匹配策略：
其中名称前带 * 的参数为必填参数，名称前未带 * 的参数为可选参数。

数据来源配置完成后，单击右下角下一步按钮，进行方案的数据缓存配置。

4.2.3 数据缓存配置

整库实时同步解决方案，目前支持“使用缓存”、“无需缓存，直接同步”这两种缓存配置方式：

• 使用缓存：
  可通过使用 Kafka 缓存来采集源端数据，这种方式需要额外配置 Kafka 数据源，并且已为对应的 Kafka 实例，创建了相应的 CDC 数据订阅采集任务，将源端 MySQL 中的数据，通过数据库传输服务中数据订阅方式，实时采集到 Kafka 实例中。在数据来源配置时绑定对应的 CDC 采集数据进入的 Kafka。数据订阅操作详见数据库传输服务。

• 无需缓存，直接同步：
  通过直接采集 MySQL Binlog 日志，进行数据实时读取。

  说明

  火山引擎中 RDS 云数据库已默认开启 Binlog 协议，若源端为其他云数据库或自建数据库，您需确认数据库是否已开启 Binlog。

本次实例中，以直接同步 Binlog 日志的方式进行实时同步，确认数据缓存同步方式后，单击右下角下一步按钮，进入数据目标设置。

4.2.4 数据目标设置

在数据目标配置界面中，完成数据源目标端设置与映射相关配置：
其中名称前带 * 的参数为必填参数，名称前未带 * 的参数为可选参数。

数据目标配置完成后，单击右下角下一步按钮，进行方案运行参数配置。

4.2.5 运行配置

1. 离线全量同步

   其中名称前带 * 的参数为必填参数，名称前未带 * 的参数为可选参数。

   配置项	说明
   *离线集成任务资源组	下拉选择 DataLeap 项目控制台中已绑定的独享数据集成资源组： 选择的资源组，需要确保能与源端、目标端数据源连通，您可单击右侧的连通性测试按钮，进入测试窗口，单击连通性测试按钮，测试选择的资源组是否可以和两侧数据源连通成功。 您也可以单击资源组管理按钮，前往资源组管理界面进行资源组的查看或新建等操作，详见资源组管理。
   *默认 Quota 数	设置可同时提交执行的集成任务数量，可根据独享集成资源组规格进行配置，如资源组的大小为 40CU，则 Quoto 配置需必须小于 20（40/2），否则会因资源问题导致任务执行时异常。
   *期望最大并发数	设置离线任务同步时，可以从源端并行读取或并行写入目标端的最大线程数。 并发数影响数据同步的效率，并发设置越高对应资源消耗也越多，由于资源原因或者任务本身特性等原因，实际执行时并发数可能小于等于设置的期望最大并发数。
   脏数据设置	您可以在此通过以下两种方式设置脏数据的处理方式： 勾选忽略所有脏数据，任务正常运行，脏数据不写入目标端。 设置脏数据的最大容忍条数或占比情况，如果配置 0，则表示不允许脏数据存在，任务会运行失败退出。
   集成高级参数设置	打开高级参数输入按钮，根据实际业务要求，以 Key\Value 形式，在编辑框中输入离线任务所需的高级参数。支持参数详见高级参数。

2. 实时增量同步
   设置解决方案中实时增量任务的运行参数情况。
   其中名称前带 * 的参数为必填参数，名称前未带 * 的参数为可选参数。

   配置项	说明
   *实时集成任务资源组	下拉选择 DataLeap 项目控制台中已绑定的独享数据集成资源组，选择的资源组，需要确保能与源端、目标端数据源连通，您可单击右侧的连通性测试按钮，进入测试窗口，单击连通性测试按钮，测试选择的资源组是否可以和两侧数据源连通成功。
   *资源设置	可通过自定义和默认两种设置方式，进行实时任务运行资源的设定，如单TaskManager CPU数量、单TaskManager内存大小、JobManager CPU数量等。 说明 默认设置中，各运行资源设置如下： 单 TaskManager CPU 数：2 单 TaskManager 内存：4096 MB 单 TaskManager slot 数：4 JobManager CPU 数：1 JobManager 内存：2048 MB
   集成高级参数设置	打开高级参数输入按钮，根据实际业务要求，以 Key\Value 形式，在编辑框中输入实时任务所需的高级参数。支持参数详见高级参数。
   Flink 运行参数设置	支持输入 Flink 相关的动态参数和执行参数，具体参数设置详见 Flink 官方文档。

3. 调度设置
   选择全量离线任务运行时所使用的调度资源组，主要用来支持离线全量任务执行时，将任务下发分配至独享数据集成资源组中运行，即离线集成任务需调度资源组和独享数据集成资源组来共同承载任务的运行。

   目前调度资源组仅支持选择公共调度资源组。

4. 提交方案

   方案运行配置完成后，单击右下角提交方案按钮，进行方案的提交，在弹窗中，勾选方案立即执行，并单击确定按钮，完成实时数据同步解决方案的创建。

5 执行详情查看

方案创建完成后，进入到方案列表界面，便可查看方案的执行概况。

1. 单击运维操作列的执行详情按钮，进入解决方案执行详情界面：

   

2. 在执行详情界面，等待所有步骤执行完成，您可根据步骤完成情况，查看各个步骤的执行详情：

   1. 您可在“一次性全量批任务启动”步骤执行详情中，查看批式任务的运维与执行日志情况。批式任务运维操作详见离线任务运维。

      

   2. 在“增量流任务启动”步骤详情中，查看增量流式任务的运维、执行日志和 Application Url 等任务详情。

      

   3. 单击更多 > Application Url 按钮，进入查看增量流式任务的执行详情，您可在该界面中看到实时同步成功数据量、失败数据量、数据量大小等详情。

      

6 数据验证

一次性全量批任务和增量流任务执行成功后，您可前往 Doris 集群目的端数据库中，验证数据的实时接收情况。

1. 前往 EMR 控制台，并登录相应的 Doris 集群。详见登录集群。

2. 登录到 Doris 集群后，执行以下语句，进入到 Doris 数据库中：

   说明

   Doris 数据库用户名密码信息和其余更多操作，详见 Doris 基础使用说明。

   mysql -h 127.0.0.1 -P9030 -u root -p
   

3. 执行以下语句，进入目的端数据库并查询 mysql_2_doris_cdc 表数据:

   use dts_test_do_not_delete;
   select * from mysql_2_doris_cdc;
   

   

   可以在 Doris 表中看到历史的全量数据。

4. 登录 MySQL 数据库，继续往数据库中插入以下数据：

   insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                                Tinytext_Info)
   values (6, '数据治理', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
           -19505.969, '2023-03-27 14:24:22',
           'KKKKKKKKKK');
   
   insert into mysql_2_doris_cdc (Id, Name, Address, Event_Time, Price, Num_Big, Num_Float, Datetime_Info,
                                Tinytext_Info)
   values (7, '数据地图', '六盘水市安次荆门路i座_282720', 1679897962, 87496.4557384283, 59534049432157,
           -19505.969, '2023-03-27 14:24:22',
           'KKKKKKKKKK');
   

5. MySQL 数据插入执行完成后，您便可在 Doris 数据库表中，查询到实时插入的数据：

   

7 后续步骤

数据验证无误后，您可以对已创建的任务进行管理，如同步方案的编辑、重启、暂停、添加任务监控等操作，方案编辑操作时，您也可对方案进行加减表、修改运行资源等操作。详见：实时整库同步解决方案管理。