多数据源一站式入湖

简介： 通过一站式入湖，将不同数据源的数据统一归并到以OSS对象存储为基础架构的集中式数据湖存储中，解决了企业面临的数据孤岛问题，为统一的数据分析打好了基础.

背景

数据湖作为一个集中化的数据存储仓库，支持的数据类型具有多样性，包括结构化、半结构化以及非结构化的数据，数据来源上包含数据库数据、binglog增量数据、日志数据以及已有数仓上的存量数据等.数据湖能够将这些不同来源、不同格式的数据集中存储管理在高性价比的存储如OSS等对象存储中，并对外提供统一的数据分析方式，有效解决了企业中面临的数据孤岛问题，同时大大降低了企业存储和使用数据的成本.

由于数据湖数据来源的多样性，如何简单高效的将这些异构数据源的数据迁移到中心化的数据湖存储中，是数据湖构建过程面临的问题.为此，我们需要提供完善的一站式入湖的能力，解决我们面临的问题，主要包括以下几点：

支持异构数据源统一的入湖方式

提供一个简单统一的入湖方式，用户可以通过简单的页面配置实现异构数据源的入湖操作.

满足数据入湖的时效性

对于日志、binglog等类型的数据源，需要实现分钟级延迟的数据入湖能力，满足实时交互式分析场景对时效性的要求.

支持数据源的实时变更

对于数据库、TableStore Tunnel等类型的数据源，源头数据会经常发生变更，比如数据层面的update，delete等操作，甚至schema层面的字段结构变更. 需要利用更好的数据格式来支持这类变更行为.

为此，阿里云新推出了数据湖构建（Data Lake Formation，DLF）服务，提供了完整的一站式入湖解决方案。

整体方案

数据湖构建的入湖技术方案如下图所示：

数据入湖整体上分为入湖模板、入湖引擎、文件格式以及数据湖存储四个部分：

入湖模板

入湖模板定义了常见的数据源入湖方式，目前主要包括RDS全量模板、DTS增量模板、TableStore模板、SLS模板以及文件格式转换5种模板.

用户根据不同的数据源选择相应的入湖模板，然后填写源头相关参数信息，即可完成入湖模板的创建，并提交给入湖引擎运行.

入湖引擎

入湖引擎使用了阿里云EMR团队自研的Spark Streaming SQL以及EMR Spark引擎，Streaming SQL基于Spark Structured Streaming，提供了相对完善的Streaming SQL语法，极大简化了实时计算的开发成本。对于实时增量模板，上层入湖模板部分将入湖模板翻译成Streaming SQL，然后提交Spark集群运行。我们在Streaming SQL里面扩展了Merge Into语法来支持update、delete操作。对于RDS等全量模板，则直接翻译成Spark SQL运行。

文件格式

DLF支持的文件格式包括Delta Lake、Parquet、json等，更多文件格式比如Hudi也在接入中。Delta Lake和Hudi等文件格式能很好的支持update、delete等操作，同时支持schema merge功能。可以很好的解决数据源实时变更问题。

数据湖存储

数据湖数据统一放在OSS对象存储中，OSS提供了海量数据存储的能力，同时在可靠性，价格等方面更具优势.

一站式入湖方案在很好的解决了前面提的几个问题：

支持异构数据源统一的入湖方式

通过模板配置，实现了统一简单的数据入湖方式.

满足数据入湖的时效性

通过自研Streaming SQL实现了分钟级延迟的数据实时入湖，满足了时效性要求.

支持数据源的实时变更

通过引进Delta Lake等更优的文件格式，实现了对update、delete等数据实时变更要求.

实时入湖

随着大数据的不断发展，用户对数据时效性的要求越来越高，实时入湖也是我们重点关注的场景，目前我们已经支持了DTS、TableStore以及SLS的实时入湖能力.

DTS增量数据实时入湖

DTS是阿里云提供了高可靠的数据传输服务，支持不同类型数据库增量数据的订阅和消费。我们实现了DTS实时订阅数据的入湖，支持用户已有订阅通道入湖和自动创建订阅通道入湖两种方式，减少用户配置成本。

在技术上，支持增量数据对历史数据的update、delete变更操作，实现分钟级延迟的数据变更感知能力。技术实现上在Streaming SQL中扩展了merge into语法来对接底层文件格式Delta Lake的相关接口.

MERGE INTO delta_tbl AS target
USING (select recordType, pk, ...from {{binlog_parser_subquery}}
) AS source
ON target.pk = source.pk
WHEN MATCHED AND source.recordType='UPDATE' THEN
UPDATE SET *
WHEN MATCHED AND source.recordType='DELETE' THEN
DELETE
WHEN NOT MATCHED THEN
INSERT *

和传统数仓的binlog入仓相比，基于数据湖的方案具有更大的优势. 在传统数仓中，为了实现数据库等变更数据的入仓，通常需要维护两张表，一张增量表用于存放每天新增的数据库变更明细数据，另外一张全量表，存放历史所有的merge数据，全量表每天和增量表更据主键做merge操作. 显然，基于数据湖方案在实现的简单性和时效性上都更优.

TableStore实时入湖

TableStore是阿里云提供的是阿里云自研的NoSQL多模型数据库，提供海量结构化数据存储以及快速的查询和分析服务.它同时支持了通道功能，支持变更数据的实时消费。我们支持TableStore全量通道、增量通道以及全量加增量通道的实现入湖.其中全量通道包含历史全量数据，增量通道包含增量变化的数据，全量加增量通道则包含了历史全量和增量变化的数据.