数据集成

概述

数据集成是稳定高效、弹性伸缩的数据同步平台，致力于提供复杂网络环境下、丰富的异构数据源之间高速稳定的数据移动及同步能力。

支持的数据源

数据集成包括离线同步和实时同步，本文为您介绍离线和实时同步支持的数据源类型。

1.离线同步支持数据源

2.实时同步支持数据源

离线数据集成配置

新建数据源

1. 选择数据源管理页面点击新建数据源。
2. 填写数据源配置信息。

1. 填写参数完毕后，点击右上角测试连接进行数据平台与将连接数据库进行连通性测试。

2. 点击测试成功后，如果是MYSQL类型数据源可点击右上角自动探知，进行查看，同步数据库下对应表模型信息。

3. 展示列表中，已勾选模型表示已创建源数据模型，可对未勾选模型进行创建源数据模型至源模型管理界面查看。

4. 点击页面右侧编辑和删除可对已创建数据源进行管理。

注意：

1. 删除数据源时，如该数据源存在关联源数据模型及工作流，该数据源无法删除；
2. INTERNAL(内部)/EXTERNAL(外部)区别：

• 面向对象：INTERNAL主要针对企业内部系统数据源（已废弃），EXTERNAL针对企业外部系统数据源。
• 功能：内部数据源在数据平台中创建的内部数据模型对应数据库有相同的表随之创建，外部数据源对应外部数据模型只拥有读取权限；

新建源数据模型

1. 进入源模型管理页面选择创建INTERNAL(内部)/EXTERNAL(外部)源数据模型。源数据模型的类型取决去数据源的类型，即INTERNAL(内部)数据源对应INTERNAL(内部)源数据模型，EXTERNAL(外部)数据源对应EXTERNAL(外部)源数据模型；
2. 点击右上角开始新增源数据模型（非mysql类型可以新建，mysql类型的直接通过数据源探知获取）；

1. 自定义模式可新增模型中字段信息，也可通过右侧编辑和删除；

2. SQL编译模式可写SQL对源数据模型进行定义表结构；

create table if not exists member (
ID id ,
PROFILE3 keyword comment '',
PROFILE19 keyword comment '',
PHONE keyword comment '',
PROFILE45 keyword comment '',
PROFILE80 keyword comment '',
OBJECTID keyword comment '',
tagCode502_VALUE keyword comment '',
primary key (id)

1. 点击页面右侧编辑和删除可对已创建源数据模型进行管理；

注意：

1. 在内部模型中仅支持ALTER TABLE语法。

2. 如果在SQL编译模式执行结果中成功则表示创建成功，若失败，可以在下方日志中查看SQL报错信息进行错误排查。

3. 操作演示。

新建数据集成任务节点

1. 离线数据集成任务配置时，任务节点所属工作流周期选择单次任务或周期任务。

2. 拖拽锚点可创建新的节点任务，选择数据集成。

3. 填写数据集成节点基础信息

1. 填写配置信息参数

1. 完成基础信息和配置信息参数填写后，可在左下角进行指标配置字段配置，左侧字段可选择性移动到右侧ods表中。

2. 移动到右侧字段表示ods表所有的字段信息，可点击最右侧功能进行字段信息的配置。

3. 可对每一个字段进行质量规则的配置，质量规则详情参考上述质量规则操作。

4. 最后点击确认按钮完成数据集成节点配置。

5. 操作演示。

实时数据集成配置

1. 实时任务配置时，任务节点所属工作流周期选择实时任务。

2. 拖拽锚点可创建新的节点任务，选择数据集成，目前实时数据集成仅支持自定义编辑操作。

3. 实时数据集成采用flink-connector，书写语法及参数配置可参考flink官网。

4. 操作演示。

-- source
CREATE TABLE `rt_test1`
(
    `freezeQty`                 DOUBLE COMMENT '冻结库存 '
    ,`distributionCode`         STRING COMMENT '分区编码 '
    ,`id`                       BIGINT
) with (
'connector' = 'mysql-cdc',
'hostname' = 'rm-bp1p4wb6181in436c33150.mysql.rds.aliyuncs.com',
'port' = '3306',
'username' = 'sync_binlog',
'password' = 'I5bWu6G515U9eX0gerpWeS8p6kJ0F461',
'database-name' = 'Demo_test',
'table-name' = 'rt_test1'
);
-- sink
CREATE TABLE `ods_rt_test1`
(
    `freezeQty`                DOUBLE COMMENT '冻结库存 ',
    `distributionCode`         STRING COMMENT '分区编码 ',
    `id`                       BIGINT,
     PRIMARY KEY (id) NOT ENFORCED
) with (
    'connector' = 'upsert-kafka',
    'topic' = 'ods_rt_test1',
    'properties.bootstrap.servers' = 'xxx.xxx.xxx.xxx:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);
-- exec
insert into ods_rt_test1
select *
from rt_test1;