阿里云云原生数据湖体系全解读——数据湖构建 数据导入_阿里云sls 样本池

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

由于数据湖数据来源的多样性， 如何简单高效的将这些异构数据源的数据迁移到中心化的数据湖存储中， 是数据湖构建过程面临的问题.为此， 我们需要提供完善的一站式入湖的能力， 解决我们面临的问题， 主要包括以下几点：
 支持异构数据源统一的入湖方式：提供一个简单统一的入湖方式， 用户可以通过简单的页面配置实现异构数据源的入湖操
作。
 满足数据入湖的时效性：对于日志、 binglog 等类型的数据源， 需要实现分钟级延迟的数据入湖能力， 满足实时交互式 分析场景对时效性的要求。
 支持数据源的实时变更：对于数据库、 TableStore Tunnel 等类型的数据源， 源头数据会经常发生变更， 比如数据层面的 update， delete 等操作， 甚至 schema 层面的字段结构变更. 需要利用更好的数据格式来支持这类变更行为。

多数据源一站式入湖

整体方案

数据入湖整体上分为入湖模板、 入湖引擎、 文件格式以及数据湖存储四个部分。入湖模板定义了常见的数据源入湖方式， 目前主要包括 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 提供了海量数据存储的能力， 同时在可靠性， 价格等方面更具优势。

实时入湖

随着大数据的不断发展， 用户对数据时效性的要求越来越高， 实时入湖也是我们重点关
注的场景， 目前我们已经支持了 DTS、 TableStore 以及 SLS 的实时入湖能力。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 *
1
2
3
4
5
6
7
8
9
10
11
12

和传统数仓的 binlog 入仓相比， 基于数据湖的方案具有更大的优势. 在传统数仓中，为了实现数据库等变更数据的入仓， 通常需要维护两张表， 一张增量表用于存放每天新增的数据库变更明细数据， 另外一张全量表， 存放历史所有的 merge 数据， 全量表每天和增量表更据主键做 merge 操作. 显然， 基于数据湖方案在实现的简单性和时效性上都更优。
TableStore 是阿里云提供的是阿里云自研的 NoSQL 多模型数据库， 提供海量结构化数据存储以及快速的查询和分析服务.它同时支持了通道功能， 支持变更数据的实时消费。我们支持 TableStore 全量通道、 增量通道以及全量加增量通道的实现入湖.其中全量通道包含历史全量数据， 增量通道包含增量变化的数据， 全量加增量通道则包含了历史全量和增量变化的数据。
SLS 是阿里云提供的针对日志类数据的一站式服务， 主要存放用户日志数据。 将 SLS中的日志数据实时归档到数据湖中， 进行分析处理可以充分挖掘数据中的价值。 目前通过SLS 入湖模板， 填写 project、 logstore 等少量信息， 即可完成日志实时入湖的能力。

数据湖构建服务+Delta Lake=CDC实时入湖

Change Data Capture(CDC)用来跟踪捕获数据源的数据变化， 并将这些变化同步到目标存储(如数据湖或数据仓库)， 用于数据备份或后续分析， 同步过程可以是分钟/小时/天等粒度， 也可以是实时同步。 CDC 方案分为侵入式(intrusive manner)和非倾入性(non-intrusive manner)两种。

侵入式方案直接请求数据源系统(如通过 JDBC 读取数据)， 会给数据源系统带来性能
压力。 常见的方案如下:
 最后更新时间(Last Modified)：源表需要有修改时间列， 同步作业需要指定最后修改时间参数， 表明同步某个时间点之后变更的数据。 该方法不能同步删除记录的变更， 同一条记录多次变更只能记录最后一次。
 自增 id 列：源表需要有一个自增 id 列， 同步作业需要指定上次同步的最大 id 值， 同步上次之后新增的记录行。 该方法也不能同步删除记录的变更， 而且老记录的变更也无法感知。
非侵入性一般通过日志的方式记录数据源的数据变化(如数据库的 binlog)， 源库需要开启 binlog 的功能。数据源的每次操作都会被记录到 binlog 中(如 insert/update/delete 等)，能够实时跟踪数据插入/删除/数据多次更新/DDL 操作等。

常见的CDC方案实现

开源常见的 CDC 方案实现主要有两种：Sqoop 离线同步和binlog 实时同步
sqoop 是一个开源的数据同步工具， 它可以将数据库的数据同步到 HDFS/Hive 中，支持全量同步和增量同步， 用户可以配置小时/天的调度作业来定时同步数据。sqoop 增量同步是一种侵入式的 CDC 方案， 支持 Last Modified 和 Append 模式。缺点：直接 jdbc 请求源库拉取数据， 影响源库性能，小时/天调度， 实时性不高，无法同步源库的删除操作， Append 模式还不支持数据更新操作。

binlog 日志可以通过一些工具实时同步到 kafka 等消息中间件中， 然后通过Spark/Flink 等流引擎实时的回放 binlog 到目标存储(如 Kudu/HBase 等)。缺点:Kudu/HBase 运维成本高，Kudu 在数据量大的有稳定性问题, HBase 不支持高吞吐的分析，Spark Streaming 实现回放 binlog 逻辑复杂， 使用 java/scala 代码具有一定门槛。

Streaming SQL+Delta Lake 实时入湖方案
阿里云 E-MapReduce 团队提供了一种新的 CDC 解决方案， 利用自研的 Streaming SQL 搭配 Delta Lake 可以轻松 实 现 C D C 实 时 入 湖 。 这 套 解 决 方 案 同 时 通 过 阿 里 云 最 新 发 布 的 数 据 湖 构 建（ Data Lake Formation， DLF） 服务提供一站式的入湖体验。

Spark Streaming SQL 在 Spark Structured Streaming 之上提供了 SQL 能力，降低了实时业务开发的门槛， 使得离线业务实时化更简单方便。Delta Lake 是 Databricks 开源的一种数据湖格式， 它在 parquet 格式之上， 提供了ACID 事务/元数据管理等能力， 同时相比parquet 具有更好的性能， 能够支持更丰富的数据应用场景(如数据更新/schema 演化等)。
E-MapReduce 团队在开源 Delta Lake 基础上做了很多功能和性能的优化， 如小文件合并 Optimize/DataSkipping/Zorder， SparkSQL/Streaming SQL/Hive/Presto 深度集成 Delta 等

Spark Streaming SQL 提供了 Merge Into 的语法， 搭配 Delta Lake 的实时写入能力， 可以很方便的实现 CDC 实时入湖方案。