关于数据湖:字节跳动基于Doris的湖仓分析探索实践

更多技术交换、求职机会，欢送关注字节跳动数据平台微信公众号，回复【1】进入官网交换群

Doris 简介

Doris 是一种 MPP 架构的剖析型数据库，次要面向多维分析，数据报表，用户画像剖析等场景。自带剖析引擎和存储引擎，反对向量化执行引擎，不依赖其余组件，兼容 MySQL 协定。
Apache Doris 具备以下几个特点：

• 良好的架构设计，反对高并发低延时的查问服务，反对高吞吐量的交互式剖析。多 FE 均可对外提供服务，并发减少时，线性裁减 FE 和 BE 即可反对高并发的查问申请。
• 反对批量数据 load 和流式数据 load，反对数据更新。反对 Update/Delete 语法，unique/aggregate 数据模型，反对动静更新数据，实时更新聚合指标。
• 提供了高可用，容错解决，高扩大的企业级个性。FE Leader 谬误异样，FE Follower 秒级切换为新 Leader 持续对外提供服务。
• 反对聚合表和物化视图。多种数据模型，反对 aggregate，replace 等多种数据模型，反对创立 rollup 表，反对创立物化视图。rollup 表和物化视图反对动静更新，无需用户手动解决。
• MySQL 协定兼容，反对间接应用 MySQL 客户端连贯，十分易用的数据利用对接。
  Doris 由 Frontend（以下简称 FE）和 Backend（以下简称 BE）组成，其中 FE 负责承受用户申请，编译，优化，散发执行打算，元数据管理，BE 节点的治理等性能，BE 负责执行由 FE 下发的执行打算，存储和治理用户数据。

数据湖格局 Hudi 简介

Hudi 是下一代流式数据湖平台，为数据湖提供了表格局治理的能力，提供事务，ACID，MVCC，数据更新删除，增量数据读取等性能。反对 Spark，Flink，Presto，Trino 等多种计算引擎。

Hudi 依据数据更新时行为不同分为两种表类型：

针对 Hudi 的两种表格局，存在 3 种不同的查问类型：

Doris 剖析 Hudi 数据的技术背景
在数仓业务中，随着业务对数据实时性的要求越来越高，T+ 1 数仓业务逐步往小时级，分钟级，甚至秒级演进。实时数仓的利用也越来越广，也经验了多个倒退阶段。目前存在着多种解决方案。

Lambda 架构

Lambda 将数据处理流分为在线剖析和离线剖析分为两条不同的解决门路，两条门路相互独立，互不影响。
离线剖析解决 T + 1 数据，应用 Hive/Spark 解决大数据量，不可变数据，数据个别存储在 HDFS 等零碎上。如果遇到数据更新，须要 overwrite 整张表或整个分区，老本比拟高。
在线剖析解决实时数据，应用 Flink/Spark Streaming 解决流式数据，剖析解决秒级或分钟级流式数据，数据保留在 Kafka 或定期（分钟级）保留到 HDFS 中。
该套计划存在以下毛病：

• 同一套指标可能须要开发两份代码来进行在线剖析和离线剖析，保护简单
• 数据利用查问指标时可能须要同时查问离线数据和在线数据，开发简单
• 同时部署批处理和流式计算两套引擎，运维简单
• 数据更新须要 overwrite 整张表或分区，老本高

Kappa 架构

随着在线剖析业务越来越多，Lambda 架构的弊病就越来越显著，减少一个指标须要在线离线别离开发，保护艰难，离线指标可能和在线指标对不齐，部署简单，组件繁多。于是 Kappa 架构应运而生。
Kappa 架构应用一套架构解决在线数据和离线数据，应用同一套引擎同时解决在线和离线数据，数据存储在音讯队列上。
Kappa 架构也有肯定的局限：

• 流式计算引擎批处理能力较弱，解决大数据量性能较弱
• 数据存储应用音讯队列，音讯队列对数据存储有有效性限度，历史数据无奈回溯
• 数据时序可能乱序，可能对局部对时序要求比拟严格的利用造成数据谬误
• 数据利用须要从音讯队列中取数，须要开发适配接口，开发简单

基于数据湖的实时数仓

针对 Lambda 架构和 Kappa 架构的缺点，业界基于数据湖开发了 Iceberg, Hudi, DeltaLake 这些数据湖技术，使得数仓反对 ACID, Update/Delete, 数据 Time Travel, Schema Evolution 等个性，使得数仓的时效性从小时级晋升到分钟级，数据更新也反对局部更新，大大提高了数据更新的性能。兼具流式计算的实时性和批计算的吞吐量，反对的是近实时的场景。

以上计划中其中基于数据湖的利用最广，但数据湖模式无奈撑持更高的秒级实时性，也无奈间接对外提供数据服务，须要搭建其余的数据服务组件，零碎较为简单。基于此背景下，局部业务开始应用 Doris 来承接，业务数据分析师须要对 Doris 与 Hudi 中的数据进行联邦剖析，此外在 Doris 对外提供数据服务时既要能查问 Doris 中数据，也要能减速查问离线业务中的数据湖数据，因而咱们开发了 Doris 拜访数据湖 Hudi 中数据的个性。

Doris 剖析 Hudi 数据的设计原理

基于以上背景，咱们设计了 Apache Doris 中查问数据湖格局 Hudi 数据，因 Hudi 生态为 java 语言，而 Apache Doris 的执行节点 BE 为 C ++ 环境，而 C ++ 无奈间接调用 Hudi java SDK，针对这一点，咱们有四种解决方案：

1. 实现 Hudi C++ client，在 BE 中间接调用 Hudi C++ client 去读写 Hudi 表。
   该计划须要残缺实现一套 Hudi C++ client，开发周期较长，前期 Hudi 行为变更须要同步批改 Hudi C++ client，保护较为艰难。
2. BE 通过 thrift 协定发送读写申请至 Broker，由 Broker 调用 Hudi java client 读取 Hudi 表。
   该计划须要在 Broker 中减少读写 Hudi 数据的性能，目前 Broker 定位仅为 fs 的操作接口，引入 Hudi 突破了 Broker 的定位。第二，数据须要在 BE 和 Broker 之间传输，性能较低。
3. 在 BE 中应用 JNI 创立 JVM，加载 Hudi java client 去读写 Hudi 表。
   该计划须要在 BE 过程中保护 JVM，有 JVM 调用 Hudi java client 对 Hudi 进行读写。读写逻辑应用 Hudi 社区 java 实现，能够保护与社区同步；同时数据在同一个过程中进行解决，性能较高。但须要在 BE 保护一个 JVM，治理较为简单。
4. 应用 BE arrow parquet c++ api 读取 hudi parquet base file，hudi 表中的 delta file 暂不解决。
   该计划能够由 BE 间接读取 hudi 表的 parquet 文件，性能最高。但以后不反对 base file 和 delta file 的合并读取，因而仅反对 COW 表 Snapshot Queries 和 MOR 表的 Read Optimized Queries，不反对 Incremental Queries。

综上，咱们抉择计划四，第一期实现了 COW 表 Snapshot Queries 和 MOR 表的 Read Optimized Queries，前面联结 Hudi 社区开发 base file 和 delta file 合并读取的 C ++ 接口。

Doris 剖析 Hudi 数据的技术实现

Doris 中查问剖析 Hudi 表面应用步骤非常简单。

1. 创立 Hudi 表面

建表时指定 engine 为 Hudi，同时指定 Hudi 表面的相干信息，如 hive metastore uri，在 hive metastore 中的 database 和 table 名字等。
建表仅仅在 Doris 的元数据中减少一张表，无任何数据挪动。
建表时反对指定全副或局部 hudi schema，也反对不指定 schema 创立 hudi 表面。指定 schema 时必须与 hiveMetaStore 中 hudi 表的列名，类型统一。
Example：

   CREATE TABLE example_db.t_hudi 
    ENGINE=HUDI
    PROPERTIES (
    "hudi.database" = "hudi_db",
    "hudi.table" = "hudi_table",
    "hudi.hive.metastore.uris"  =  "thrift://127.0.0.1:9083"
    );
    
    
    CREATE TABLE example_db.t_hudi (
    column1 int,
    column2 string)
    ENGINE=HUDI
    PROPERTIES (
    "hudi.database" = "hudi_db",
    "hudi.table" = "hudi_table",
    "hudi.hive.metastore.uris"  =  "thrift://127.0.0.1:9083"
    );

2. 查问 Hudi 表面

查问 Hudi 数据表时，FE 在 analazy 阶段会查问元数据获取到 Hudi 表面的的 hive metastore 地址，从 Hive metastore 中获取 hudi 表的 schema 信息与文件门路。

1. 获取 hudi 表的数据地址
2. FE 布局 fragment 减少 HudiScanNode。HudiScanNode 中获取 Hudi table 对应的 data file 文件列表。
3. 依据 Hudi table 获取的 data file 列表生成 scanRange
4. 下发 HudiScan 工作至 BE 节点
5. BE 节点依据 HudiScanNode 指定的 Hudi 表面文件门路调用 native parquet reader 进行数据读取。

前期布局

目前 Apche Doris 查问 Hudi 表已合入社区，以后已反对 COW 表的 Snapshot Query，反对 MOR 表的 Read Optimized Query。对 MOR 表的 Snapshot Query 临时还未反对，流式场景中的 Incremental Query 也没有反对。
后续还有几项工作须要解决，咱们和社区也在积极合作进行中：

1. MOR 表的 Snapshot Query。MOR 表实时读须要合并读取 Data file 与对应的 Delta file，BE 须要反对 Delta file AVRO 格局的读取，须要减少 avro 的 native 读取形式。
2. COW/MOR 表的 Incremental Query。反对实时业务中的增量读取。
3. BE 读取 Hudi base file 和 delta file 的 native 接口。目前 BE 读取 Hudi 数据时，仅能读取 data file，应用的是 parquet 的 C ++ SDK。前期咱们和联结 Hudi 社区提供 Huid base file 和 delta file 的 C ++/Rust 等语言的读取接口，在 Doris BE 中间接应用 native 接口来查问 Hudi 数据。

本文为字节跳动数据平台研发工程师在 DataFunSummit 大会演讲实录，关注字节跳动数据平台微信公众号，回复【0929】，支付本次分享 PPT。

立刻跳转火山引擎 E -MapReduce 官网理解更多信息