数据集成产品分析

Posted 2023-02-13 爱是与世界平行

• 数据集成产品分析

数据集成产品致力于异构数据源之间的数据交换与汇聚，该类产品是数据中台建设的第一环节，笔者将通过两篇文章对数据集成产品进行分析。

数据同步，又称为数据集成、数据迁移，主要用于实现不同系统间的数据流转。

为什么会有数据同步这类产品？

在企业中，业务的快速发展产生了大量数据，也催生出多种应用系统，各系统承载不同类型的数据，对应着不同的数据存储方式。

而对于构建数据中台或大数据系统，首先需要将企业内部各个业务系统的数据实现互联互通，从物理上打破数据孤岛，而这主要通过数据汇聚和同步的能力来实现。

数据同步方式有多种：API接口同步、数据文件同步和数据库日志解析同步，适用于不同的业务场景。

本次分享的数据同步是基于数据库日志解析的方式实现，其主要应用场景是：数据从业务系统同步到数仓，和数据从数仓同步到数据应用两个方面。

1 数据集成产品简介

1.1 产品介绍

数据同步致力于保证数据在不同数据源之间被高效准确地迁移。根据数据时效性要求和应用场景，数据同步可分为离线同步和实时同步：

1）离线同步

主要用于大批量数据的周期性迁移，对时效性要求不高，一般采用分布式批量数据同步方式，通过连接读取数据，读取数据过程中可以有全量、增量方式，经过统一处理后写入目标存储。

成熟的产品有：Sqoop、DataX、kettle等。

2）实时同步

针对数据时效性要求高的场景，其将源端数据的变化实时同步到目标端数据源中，保证源端数据与目标端数据实时保持一致，就可满足业务实时查询分析使用数据或实时计算等需求。

成熟的产品有：Canal、otter等。

在实际业务场景中，离线同步和实时同步搭配使用，为保证已有的数据表在目标端数据源中可用，会使用离线同步将该表的历史数据全量迁移到目标端数据源中，对于增量数据则通过实时集成来增量迁移。

1.2 核心流程

数据集成的实现有三个关键步骤：数据读取、数据转换（映射）、数据写入，核心流程如下图所示：

具体流程为：数据集成从源端数据源中读取数据，按照建好的映射关系及转换规则，将数据写入到目标数据源中。其中：

• 配置源端数据源和目标端数据源，用于连接数据源，获取数据源的读、写权限等
• 建立映射关系：源端数据源中数据表及字段对应到目标端数据源中的哪张表和字段。建立映射关系后，数据集成根据映射关系，将源端数据写入到对应的目标端数据
• 数据转换规则：数据写入时，可能有不同数据格式转换，敏感数据脱敏展示、无用字段过滤、null值处理等需求，可以制定相应的数据转换规则来实现
• 数据传输通道：可配置任务并发数、数据连接数等任务参数，达到数据集成的任务被高效稳定执行等目的

基于上述流程，数据集成产品的核心功能一般会包含以下4个：

1. 数据源配置
2. 同步任务配置
3. 任务调度
4. 监控告警

数据集成平台在进行异构数据源之间的数据迁移时，需要保证迁移任务被高效完成，被迁入目标端数据源的数据是准确可靠的、实时的，数据传输过程是安全的等，这是用户核心关注点，也是期望平台达到的目标。

1.3 其他大数据产品的联系和区别

在构建数据仓库的场景中，数据加载到数仓后，随后进行数据加工和数据应用，其中涉及的3类大数据产品如下：

1）数据集成：面向数据汇聚与交换

产品流程：配置数据源—创建同步任务—配置调度任务。

核心任务：ETL、校验、补数、合并。

2）数据加工：面向数据分析

产品流程：创建表—编写加工逻辑（insert）—配置调度任务。

数据加工方式：离线计算、实时计算、机器学习等。

3）任务调度：工作流编排

产品流程：创建任务节点—配置节点依赖关系—提交并执行。

任务调度：任务执行、任务调度与运维。

联系：

• 数据集成和数据加工都是数据生命周期中的一环
• 数据集成任务和数据加工任务其实就是任务调度中的任务节点job，任务调度保证数据被顺序采集和加工出来

以用户画像分析为例，oss_数据同步和rds_数据同步两个节点是数据集成任务节点，ods_log_info_d、dws_user_info_all_d、rpy_user_info_d三个节点是数据加工任务节点，绘制各节点间的连线即工作流编排。

提交并执行画布中的流程节点，数据就会按照箭头方向执行各节点，产出最终结果。

**区别：**数据集成和数据加工都基于任务执行和调度实现，两者虽然都是ETL，但是各自关注的重点并不相同。

①核心动作

数据集成核心动作是抽数（读、写、转换），也就是数据交换，转换动作也只是简单的数据清洗。

数据加工的核心动作是计算逻辑/清洗等加工规则的编写，加工规则复杂。

②核心目标

数据集成的核心目标，是保证数据被高效准确地迁移到目标端数据源。

数据加工的核心目标，是加工规则编写准确“翻译”业务需求。

2 数据集成产品流程

离线集成与实时集成在实际运行中，关注的侧重点是不同的，对于离线集成，面对的是批数据，更多考虑大规模数据量的迁移效率问题；对于实时集成，面对的是流数据，更多考虑数据准确性问题。

数据同步中一般采用先全量在增量的方式来进行数据的迁移，即先将历史数据全量迁移到目标数据源，完成初始化动作，在进行增量数据的实时同步，这样保证目标端数据的可用性。当然也有不care历史数据的场景，此时就无需进行全量迁移的动作。

2.1 实时集成

我们主要采用基于日志的CDC方式来实现增量数据的实时同步，CDC即change data capture，捕获数据的变化。

实时集成通过读取源端数据库日志bin_log来捕获数据的变化情况（insert、update、delete），将其传输到kafka topic中，然后通过spark streaming对数据进行转换/清洗，写入到stg增量表中，最后将增量数据与全量数据合并到数仓ods表中。

由于数据库日志抽取一般是获取所有的数据记录的变更（增、删、改），落到目标表时，需要根据主键去重，并按照日志时间倒序排列获取最后状态的变化情况。

具体的实时集成任务执行逻辑及流程如下所示：

实时集成有以下特点：

• 源端数据库产生一条记录，数据集成实时同步一条记录
• 流数据在数据传输过程中可能会被丢失或延迟

故相较于离线集成，在数据同步链路上，实时集成会增加数据校验和数据合并两个动作。

• 数据校验主要校验源端和目标端的数据量，保证数据没有被丢失
• 数据合并则是由ods库base表和cdc目标库increment表组成，具体逻辑为：将stg数据按主键去重，取最新一条，根据主键与ods数据表中的T+1数据合并

2.2 离线集成

离线集成分为全量和增量两种方式对大规模数据进行批量迁移。

• 全量迁移是将某些表的全部历史数据同步到目标数据源中
• 增量迁移，通常需要使用where子句（RDB）或者query子句（MongoDB）等增量配置参数，同时在结合调度参数（定时任务的重复周期：分钟、小时、天、周、月等）可实现增量迁移任意指定日期内的数据。

比如，想要实现每日数据的增量同步，各参数可配置为：

• Where子句：配置为DS=’$dateformat(yyyy-MM-dd,-1,DAY)’，
• 配置定时任务：重复周期为1天，每天的凌晨0点自动执行作业

增量迁移的数据可以对目标端数据源表中数据进行追加、覆盖和更新操作。

3 DataWorks-数据集成

DataWorks-数据集成提供了离线集成和实时集成两种方案，同时提供了面向不同业务场景的同步任务配置化方案，支持不同数据源的一键同步功能，进行简单的配置，就可以完成一个复杂业务场景。

• 思维导图地址：https://www.processon.com/view/link/6224c1db6376895389223dce

3.1 离线集成

1）基本介绍

离线（批量）的数据通道通过定义数据来源和去向的数据源和数据集，提供一套抽象化的数据抽取插件（Reader）、数据写入插件（Writer），并基于此框架设计一套简化版的中间数据传输格式，从而实现任意结构化、半结构化数据源之间数据传输。

离线集成支持通过向导模式和脚本模式来实现批量数据同步。支持对单表、文件、分库分表、整库（所有表/部分表）进行全量和增量迁移。

在增量同步中，主要通过where（RDBMS）、query（MongoDB）等子条件抽取源端增量数据（指定时间区间），并配合调度参数写入目标端对应的日期分区中，实现增量同步（追加、覆盖/更新）。

对于单表、多表（分库分表）迁移到单目标表时，支持全量和增量方式进行迁移（源表和目标表中的字段建立映射关系）。

对于整库（所有表/部分表）迁移到目标源时，考虑到历史数据量较大，提供了周期性全量（分批上传）、一次性全量、周期性增量、一次性增量、一次性全量周期性增量5种迁移方案（源表与目标表通过表名建立映射关系，目标端没有表时，支持自动建表）。

2）特点/优势

a.支持丰富的数据源

DataWorks-数据集成已经商业化，并且开源到社区中。因此，①支持的数据源非常丰富，②插件化

• 离线支持50+种数据源，实时支持10+数据源
• 读写任意组合（任意两种数据源组合搭配）
• 涵盖关系型数据库、MPP，NoSQL、文件存储、消息流等各大种类

b.灵活的配置

①支持向导模式和脚本模式配置数据同步任务，更方便和灵活，

• 向导模式对业务新手来说，上手门槛低
• 脚本模式可以提供更丰富灵活的能力，自定义配置同步条件，实现精细化的配置管理

比如，在某些业务场景中，where配置项不足以描述所筛选的条件，脚本模式下支持使用querySql来自定义筛选SQL，比如需要进行多表join后同步数据。

②数据同步任务场景化，基于不同的场景提供配置好的同步方案

c.高效的同步速率

离线同步中，dataworks通过下述几种方式控制/提高同步速率

①切分键

将数据表中某一列作为切分键（建议使用主键或有索引的列作为切分键），读取数据时，根据配置的字段进行数据切片，实现并发读取，可以提升数据同步速率

②通道配置

• 任务期望最大并发数

​ 数据同步任务内，可以从源并行读取或者并行写入数据存储端的最大线程数

• 同步速率

​ 设置同步速率可以保护读取端数据库，以避免抽取速度过大，给源库造成太大压力

• 分布式处理能力

​ 数据同步时，可以将任务分散到多台执行节点上并发执行，提高同步速率

​ 该模式下，配置较大任务并发数会增加数据存储访问压力

• 目标端写入并发数

③离线数据同步任务调优

3.2 实时集成

1）基本介绍

数据集成的实时同步包括实时读取、转换和写入三种基础插件，各插件之间通过内部定义的中间数据格式进行交互。

一个实时同步任务支持多个转换插件进行数据清洗，并支持多个写入插件实现多路输出功能。同时针对某些场景，支持整库实时同步解决方案，您可以一次性实时同步多个表。

以关系型数据库实时同步到maxcompute为例，

在实时同步过程中，源端表的变更记录实时写入MaCompute的是log表，然后会经过拆分（split）变为Delta表，最后Delta表再与Base表进行合并（Merge），最终结果会写入Base表中

在实际执行中，实时集成会根据base表自动生产log表和delta表，并且自动生成拆分（split）与合并（merge）的离线任务，并会自动启动他们，作为用户只需要关心最终的base表即可。

• 在写入目标目标表时，写入模式分为写入日志表和写入增量表

• • 写入日志表表示源端所有表的变更记录写入一张maxcompute log表中，只有在“同步解决方案”里才支持此选项，因为要配合着对应的拆分任务才有意义
  • 写入增量表表示源端表的变更记录直接写入对应的maxcompute增量表中，此模式目前暂不在“同步解决方案”可选，因为“同步解决方案”中已经可以自动建立上下游所有任务了，可以完整实现最终全量表的结果更新

• 实时同步在同步mysql、Oracle、loghub和polarDB类型的数据至maxcompute、datahub或kafka时，会在同步的目标端（日志表/增量表）添加5个附加列，以便进行元数据管理、排序去重等操作

• 实时同步校验数据准确性通过手动配置数据质量规则来对比数据量的一致性

2）特点/优势

• 数据源丰富多样

​ 支持星型链路组合，可以将多种输入及输出数据源搭配组成同步链路进行数据同步。

• 采用解决方案系统

​ 支持常见数据库的整库同步，实现先全量，再持续增量同步数据。

• 同步数据方式多样

​ 可以选择分库分表、单表或整库多表等多种方式进行数据同步，同时，也可以根据不同DDL消息配置不同实时同步规则。

​ 可以编辑已经配置运行的同步任务，为该任务快速添加表或删除已有同步表。

• 支持数据处理

​ 可以根据业务需求，对输入数据源进行数据过滤、字符串替换和数据脱敏处理后再进行输出。

• 支持监控运维告警

​ 对于业务延迟、Failover、脏数据、心跳检查和失败信息，支持通过邮件、短信、电话和钉钉通知发送相应告警，方便及时发现并处理报错信息。

• 使用图形化开发

​ 无需编写代码，直接通过托拽的方式即可进行任务的开发。业务新手也能够轻松上手。

4 DGC-数据集成

DGC-数据集成提供离线同步和实时同步两种同步方式。

离线同步通过向导模式实现，实时同步则通过脚本命令的方式实现。

4.1 离线同步

1）基本介绍

CDM数据迁移采用抽取-写入模式，CDM首先从源端数据源抽取数据，然后再将数据写入目标端数据源，数据访问操作都是有CDM主动发起。迁移过程中需要提供源端与目标端数据源的用户名和密码，存放在CDM实例的数据库中。对于数据源（如RDS数据源）支持SSL时，会使用SSL加密传输。

在创建数据迁移任务前，需要创建数据连接，让CDM能够读写数据源。一个迁移任务需要建立两个连接：源连接和目的连接。

2）操作流程

step1：创建CDM集群

step2：创建连接

通过创建数据连接让CDM集群能够读写数据源

需要创建两个连接，一个用于CDM连接到源端数据源，一个用于CDM连接到目的端数据源

step3：创建同步作业

• 输入作业名称
• 选择源连接、目标连接
• 配置源端与目标端作业参数

step4：字段映射

CDM会自动匹配源端和目标端数据表字段，需要用户检查目的端的数据是否完整（支持添加新字段方式补齐）、字段映射关系和时间格式是否正确（支持拖拽方式调整映射关系），比如源字段类型是否可以转换为目标字段类型

step5：配置任务参数

任务参数配置包括是否定时任务、作业失败是否重试、是否写入脏数据等

step6：保存并运行作业

3）产品功能

• 表/文件/整库迁移

支持批量迁移表或者文件，还支持同构/异构数据库之间整库迁移，一个作业即可迁移几百张表。

• • 在数据库迁移场景下，一个作业等价于迁移一张表，在文件迁移场景下，一个作业可以迁移多个文件

• 增量数据迁移

支持文件增量迁移、关系型数据库增量迁移、HBase/CloudTable增量迁移，以及使用Where条件配合时间变量函数实现增量数据迁移。

• 事务模式迁移

支持当CDM作业执行失败时，将数据回滚到作业开始之前的状态，自动清理目的表中的数据。

• • CDM会自动创建临时表（阶段表），先将数据导入到该临时表，导入成功后再通过数据库的事务模式将数据迁移到目标表中，导入到临时表失败的话，则将目标表回滚到作业开始之前的状态
  • 适用于数据库离线迁移场景，仅支持有限的数据库增量迁移，不支持数据库实时增量迁移

• 字段转换

支持去隐私、字符串操作、日期操作等常用字段的数据转换功能。

• 文件加密

在迁移文件到文件系统时，CDM支持对写入云端的文件进行加密。

• MD5校验一致性

支持使用MD5校验，检查端到端文件的一致性，并输出校验结果。

• 脏数据归档

支持将迁移过程中处理失败的、被清洗过滤掉的、不符合字段转换或者不符合清洗规则的数据单独归档到脏数据日志中，便于用户查看。并支持设置脏数据比例阈值，来决定任务是否成功。

4.2 实时同步

1）基本介绍

DGC实时数据接入服务（Data Ingestion Service）为处理或分析流数据的自定义应用程序构建数据流管道，主要解决云服务外的数据实时传输到云服务内的问题。实时数据接入每小时可从数十万种数据源（如日志和定位追踪事件、网站点击流、社交媒体源等）中连续捕获、传送和存储数TB数据。

实时集成支持的数据源：

2） 操作流程

step1：购买实时数据接入通道，开通DIS通道

step2：准备DIS应用开发环境

用户开发DIS应用程序前，首先需要安装应用开发工具。然后获取SDK和样例工程，并导入到用户的开发环境中

step3：发送数据到接入通道

基于数据上传业务安装应用程序，并运行程序，实现数据上传功能。数据上传过程中可在Console控制台查看数据上传通道相关信息

step4：从DIS获取数据

基于删除下载业务开发应用程序，并运行程序，实现数据下载功能