DataSail 全域数据集成提供流式数据同步能力，支持丰富的异构数据源之间，进行高速稳定的实时数据同步能力，将源端数据库中数据实时同步至目标数据库中，实现目标库数据实时保持和源库的数据对应。
本文将为您介绍流式数据同步的操作说明。

1 数据源

数据集成使用的数据源，需要先在项目控制台下注册，数据集成同步支持的数据源类型，详见支持的数据源。
数据源需要在网络连通的前提下进行数据同步，相关说明详见配置网络连通 。

2 新建流式数据集成

创建流式数据集成任务的步骤如下：

1. 登录 DataLeap 租户控制台。
2. 在左侧导航栏，单击项目管理，进入项目列表界面。
3. 单击相应的项目名称，进入到数据开发界面。
   
4. 在数据开发界面，单击目录树上新建任务按钮，进入新建任务界面。
5. 选择任务类型：
   1. 分类：数据集成。
   2. 选择任务：流式集成。
6. 填写任务基本信息：
   1. 任务名称：输入任务的名称，只允许字母、数字、下划线和连字符，且需要在63个字符以内。如：stream_0101-test
   2. 保存至： 选择任务存放的目标文件夹目录。
7. 单击确定按钮，完成任务创建。

3 任务配置

3.1 任务配置

新建流式集成任务完成后，需要配置相关任务信息，流式集成任务已支持可视化和脚本模式配置，操作步骤如下：

3.1.1 可视化配置

1. 选择数据源。

   1. 选择数据来源信息，支持选择 BMQ、DataSail、Kafka、RocketMQ、TDengine、TLS 数据源类型。
   2. 选择需写入的目标数据源信息，支持选择 ByteHouse_CE、CFS、ClickHouse、Doris、Elasticsearch 等十余种数据源类型。

   说明

   • 各数据源配置说明，详见数据源列表。
   • 部分数据源写入时，需要选择数据写入方式。针对不同的数据源，有不同的写入方式。

   

2. 配置来源端和目标端字段的映射关系。
   字段映射支持选择基础模式和转换模式配置映射：

   注意

   基础模式和转换模式不支持互相切换，模式切换后，将清空现有字段映射中所有配置信息，一旦切换无法撤销，需谨慎操作。

   • 转换模式：
     字段映射支持数据转换，您可根据实际业务需求进行配置，将源端采集的数据，事先通过数据转换后，以指定格式输入到目标端数据库中。
     转换模式详细操作说明详见4.1 转换模式
     在转换模式中，你可依次配置：来源节点、数据转换、目标节点信息：

     配置节点	说明
     来源节点	配置数据来源 Source 节点信息： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 数据字段：通过自动添加、手动添加等方式添加数据来源字段信息。 配置完成后，单击确认按钮，完成来源节点配置。
     数据转换	单击数据转换右侧添加按钮，选择 SQL 转换方式，配置转换信息和规则： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 SQL 脚本：输入 SQL 脚本转换规则，目前仅支持添加一个转换的 SQL 语句，且不能包括 “;”。 配置完成后，单击确认按钮，完成数据转换节点配置。SQL 脚本示例详见4.1.2 添加转换节点。
     目标节点	配置目标节点 Sink 信息： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 数据字段：通过自动添加、手动添加等方式添加数据目标字段信息。 配置完成后，单击确认按钮，完成目标节点配置。

   • 基础模式：

     • 单击自动添加，可以根据数据源 Schema 自动添加字段。部分数据源类型支持获取schema自动添加字段。
     • 单击手动添加，可以添加一条空白的字段信息，手工输入字段名、类型等。
     • 单击同名映射，可根据目标端/源端的字段添加情况，进行同名映射添加源端/目标端的字段信息。

       说明

       同名映射，仅部分数据源支持，如云原生消息引擎 BMQ 数据源。

     • 单击删除全部，可以删除全部字段信息。
     • 通过拖拽字段左侧的按钮，可以调整字段顺序。

     注意

     请注意列与列之间映射的字段类型是否数据兼容。

     

3. 设置以下任务运行参数。

   • 是否开启归档：选择开启归档后，可指定归档字段。
   • 默认消费起始：当前任务启动时，对 Kafka 的默认消费位置。

     说明

     Kafka/Bmq/RocketMQ 数据源类型的流式集成任务在流式运维界面重启任务操作时，也可按需开启“重置 offset”操作，并指定最新、最老、时间戳或分区 offset 形式的消费起始位置。操作详见流式运维-单任务操作-重启。

   • 高级参数：选择开启时，需要设置高级参数的 Key 和 value 值，可通过单行编辑模式和脚本编辑模式添加，例如以单行编辑模式添加流式集成任务的全局并发参数，您可参考以下配置：
     

   说明

   流式集成作业的默认全局并发为 MQ Partition 数量/4，若 MQ 流量较小时，可直接在任务高级参数中新增参数：job.common.global_parallelism_num = ${并发数}，通过这种方式来指定任务全局并发，该参数最终决定会拉起多少 TaskManager 数量；若 Partition 数量很多，但整体流量不高的话，您也可以通过指定该参数，来节约任务实际执行资源。

3.1.2 脚本模式配置

1. 进入流式集成任务配置界面，单击上方工具栏中的切换至脚本模式按钮，进入脚本配置界面。

   注意

   可视化模式切换至脚本模式，将清空可视化模式中现有的配置，且一旦切换无法撤销。

   
2. 单击工具栏中的导入脚本模板按钮，选择脚本模板来源类型、目标类型、模板库等信息：
   
3. 选择流式任务模板后，您便可根据实际数据源信息，进行脚本修改。各数据源脚本配置说明，详见数据源配置。

3.2 设置参数

数据源相关任务配置完成后，可单击右侧的参数设置 ，进入该流式任务的参数设置页面，可以设置流式任务的基础信息。

3.3 配置数据集成资源组

参数配置完成后，可单击右侧的数据集成资源组配置按钮，进入数据集成资源组页面。
下拉选择在项目控制台中，已完成资源组连通性测试的数据集成资源组，保障任务网络能够连通。更多说明请参见：数据集成资源组。

4 提交任务

任务所需参数配置完成后，将任务提交发布到运维中心实时任务运维中执行。 单击操作栏中的保存和提交上线按钮，在弹窗中，需先通过提交上线流程，最后单击确认按钮，完成作业提交。详见5.2 流式任务提交。
后续任务运维操作详见：实时任务运维。