上海网站建设兴策,dremrever怎么做网站,找客户的十大方法,as.net 网站开发视频教程Apache Iceberg 概述和源代码的构建 1.数据湖的解决方案 - Iceberg1.1 Iceberg 是什么1.2 Iceberg 的 Table Format 介绍1.3 Iceberg 的核心思想1.4 Iceberg 的元数据管理1.5 Iceberg 的重要特性1.5.1 丰富的计算引擎1.5.2 灵活的文件组织形式1.5.3 优化数据入湖流程1.5.4 增量… Apache Iceberg 概述和源代码的构建 1.数据湖的解决方案 - Iceberg1.1 Iceberg 是什么1.2 Iceberg 的 Table Format 介绍1.3 Iceberg 的核心思想1.4 Iceberg 的元数据管理1.5 Iceberg 的重要特性1.5.1 丰富的计算引擎1.5.2 灵活的文件组织形式1.5.3 优化数据入湖流程1.5.4 增量读取处理能力 1.6 数据文件结构1.6.1 元数据文件1.6.1.1 Table Metadata1.6.1.2 快照Snapshot1.6.1.3 清单文件Manifest File 1.6.2 数据文件 2.Apache Iceberg 的实现细节2.1 快照设计方式2.1.1 快照隔离2.1.2 增量读取数据2.1.3 原子性操作 2.2 事务性提交2.2.1 写操作要求2.2.2 冲突解决 - 乐观锁 3.Iceberg 结合 Flink 场景分享3.1 构建近实时 Data Pipeline3.2 CDC 数据实时摄入摄出3.3 从 Iceberg 历史数据启动 Flink 任务3.4 通过 Iceberg 数据来修正实时聚合结果 4.Iceberg 0.11.1 源代码编译4.1 编译 Iceberg4.1.1 下载 Iceberg 0.11.1 软件包4.1.2 解压 Iceberg 0.11.1 软件包4.1.3 修改对应的版本4.1.4 编辑 build.gradle 文件添加国内源4.1.5 下载依赖可选4.1.6 正式编译4.1.7 生成的目录 4.2 Iceberg 环境部署 5.总结 我们在使用不同的引擎进行大数据计算时需要将数据根据计算引擎进行适配。这是一个相当棘手的问题为此出现了一种新的解决方案介于上层计算引擎和底层存储格式之间的一个中间层。这个中间层不是数据存储的方式只是定义了数据的元数据组织方式并向计算引擎提供统一的类似传统数据库中 “表” 的语义。它的底层仍然是 Parquet、ORC 等存储格式。
基于此Netflix 开发了 Iceberg目前已经是 Apache 的顶级项目https://iceberg.apache.org/。
1.数据湖的解决方案 - Iceberg
1.1 Iceberg 是什么 Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to compute engines including Flink Trino Spark and Hive using a high-performance table format that works just like a SQL table. Iceberg 是一种开放的数据湖表格式。可以简单理解为是基于计算层Flink、Spark和存储层ORCParquetAvro的一个中间层用 Flink 或者 Spark 将数据写入 Iceberg然后再通过其他方式来读取这个表比如 SparkFlinkPresto 等。 在文件 FormatParquet / Avro / ORC 等之上实现 Table 语义
支持定义和变更 Schema支持 Hidden Partition 和 Partition 变更ACID 语义历史版本回溯借助 Partition 和 Columns 统计信息实现分区裁剪不绑定任何存储引擎可拓展到 HDFS / S3 / OSS 等容许多个 writer 并发写入乐观锁机制解决冲突
1.2 Iceberg 的 Table Format 介绍
Iceberg 是为分析海量数据而设计的被定义为 Table FormatTable Format 介于计算层和存储层之间。
Table Format 向下管理在存储系统上的文件向上为计算层提供丰富的接口。存储系统上的文件存储都会采用一定的组织形式譬如读一张 Hive 表的时候HDFS 文件系统会带一些 Partition、数据存储格式、数据压缩格式、数据存储 HDFS 目录的信息等这些信息都存在 Metastore 上Metastore 就可以称之为一种文件组织格式。
一个优秀的 文件组织格式如 Iceberg可以更高效的支持上层的计算层访问磁盘上的文件做一些 list、rename 或者查找等操作。
表和表格式是两个概念。表是一个具象的概念应用层面的概念我们天天说的表是简单的行和列的组合。而 表格式 是数据库系统实现层面一个抽象的概念它定义了一个表的 Scheme 定义包含哪些字段表下面文件的组织形式Partition 方式、元数据信息表相关的统计信息表索引信息以及表的读写 API如下图左侧所示 上图右侧是 Iceberg 在数据仓库生态中的位置和它差不多相当的一个组件是 Metastore。不过 Metastore 是一个服务而 Iceberg 就是一系列 jar 包。对于 Table Format我认为主要包含 4 4 4 个层面的含义分别是 表 Schema 定义是否支持复杂数据类型表中文件的组织形式表相关统计信息、表索引信息以及表的读写 API 信息。
表 Schema 定义了一个表支持字段类型比如 int、string、long 以及复杂数据类型等。表中文件组织形式最典型的是 Partition 模式是 Range Partition 还是 Hash Partition。Metadata 数据统计信息。表的读写 API。上层引擎通过对应的 API 读取或者写入表中的数据。
1.3 Iceberg 的核心思想 Iceberg 的核心思想就是 在时间轴上跟踪表的所有变化
快照 表示表数据文件的一个完整集合。每次更新操作会生成一个新的快照。
1.4 Iceberg 的元数据管理 从图中可以看到 Iceberg 将数据进行分层管理主要分为 元数据管理层 和 数据存储层。元数据管理层又可以细分为三层
Metadata FileSnapshotManifest
Metadata File 存储当前版本的元数据信息所有 Snapshot 信息Snapshot 表示当前操作的一个快照每次 commit 都会生成一个快照一个快照中包含多个 Manifest。每个 Manifest 中记录了当前操作生成数据所对应的文件地址也就是 data files 的地址。基于 Snapshot 的管理方式Iceberg 能够进行 time travel历史版本读取以及增量读取并且提供了 serializable isolation。
数据存储层支持不同的文件格式目前支持 Parquet、ORC、AVRO。
1.5 Iceberg 的重要特性
Apache Iceberg 设计初衷是 为了解决 Hive 离线数仓计算慢的问题经过多年迭代已经发展成为构建数据湖服务的表格式标准。关于 Apache Iceberg 的更多介绍请参见 Apache Iceberg 官网。
目前 Iceberg 提供以下核心能力 1.5.1 丰富的计算引擎
优秀的内核抽象使之不绑定特定引擎目前在支持的有 Spark、Flink、Presto、Hive。Iceberg 提供了 Java Native API不用特定引擎也可以访问 Iceberg 表。
1.5.2 灵活的文件组织形式
提供了 基于流式的增量计算模型 和 基于批处理的全量表计算模型批任务和流任务可以使用相同的存储模型HDFS、OZONE数据不再孤立以构建低成本的轻量级数据湖存储服务。Iceberg 支持隐藏分区Hidden Partitioning和分区布局变更Partition Evolution方便业务进行数据分区策略更新。支持 Parquet、ORC、Avro 等存储格式。
1.5.3 优化数据入湖流程
Iceberg 提供 ACID 事务能力上游数据写入即可见不影响当前数据处理任务这大大简化了 ETL。Iceberg 提供 Upsert / Merge Into 行级别数据变更可以极大地缩小数据入库延迟。
1.5.4 增量读取处理能力
Iceberg 支持通过流式方式读取增量数据实现主流开源计算引擎入湖和分析场景的完善对接。支持 Spark Structed Streaming。支持 Flink Table Source。支持历史版本回溯。
1.6 数据文件结构
我们先了解一下 Iceberg 在文件系统中的布局总体来讲 Iceberg 分为两部分数据。
第一部分是 数据文件如下图中的 .parquet 文件。第二部分是 表元数据文件Metadata 文件包含 Snapshot 文件snap-*.avro、Manifest 文件.avro、TableMetadata 文件*.json等。 1.6.1 元数据文件
其中 Metadata 目录存放元数据管理层的数据表的元数据是不可修改的并且始终向前迭代当前的快照可以回退。
1.6.1.1 Table Metadata
version[number].metadata.json存储每个版本的数据更改项。
1.6.1.2 快照Snapshot
snap-[snapshotID]-[attemptID]-[commitUUID].avro存储快照 Snapshot 文件。
快照代表一张 Iceberg 表在某一时刻的状态也被称为 清单列表Manifest List里面存储的是清单文件列表每个清单文件占用一行数据。清单列表文件以 snap 开头以 avro 后缀结尾每次更新都产生一个清单列表文件。每行中存储了清单文件的路径。
清单文件Manifest Files里面存储数据文件的分区范围、增加了几个数据文件、删除了几个数据文件等信息。数据文件Data Files存储在不同的 Manifest Files 里面Manifest Files 存储在一个 Manifest List 文件里面而一个 Manifest List 文件代表一个快照。
1.6.1.3 清单文件Manifest File
[commitUUID]-[attemptID]-[manifestCount].avroManifest 文件。
清单文件是以 avro 格式进行存储的以 avro 后缀结尾每次更新操作都会产生多个清单文件。其里面列出了组成某个快照Snapshot的数据文件列表。每行都是每个数据文件的详细描述包括 数据文件的状态、文件路径、分区信息、列级别的统计信息比如每列的最大最小值、空值数等、文件的大小 以及 文件里面数据的行数 等信息。其中列级别的统计信息在 Scan 的时候可以为算子下推提供数据以便可以过滤掉不必要的文件。
1.6.2 数据文件
data 目录组织形式类似于 Hive都是以分区进行目录组织图中 dt 为分区列。
Iceberg 的数据文件通常存放在 data 目录下。一共有三种存储格式Avro、ORC 和 Parquet主要是看你选择哪种存储格式后缀分别对应 avro、orc 或者 parquet。在一个目录通常会产生多个数据文件。
2.Apache Iceberg 的实现细节
2.1 快照设计方式
2.1.1 快照隔离
读操作仅适用当前已生成快照。写操作会生成新的隔离快照并在写完成后原子性提交。
如下图所示虚线框Snapshot-1表示正在进行写操作但是还没有发生 commit 操作这时候 Snapshot-1 是不可读的用户只能读取已经 commit 之后的 Snapshot。同理Snapshot-2Snapshot-3 表示已经可读。 可以支持并发读例如可以同时读取 S1、S2、S3 的快照数据同时可以回溯到 Snapshot-2 或者 Snapshot-3。在 Snapshot-4 commit 完成之后这时候 Snapshot-4 已经变成实线就可以读取数据了。
例如现在 Current Snapshot 的指针移到 S3用户对一张表的读操作都是读 Current Snapshot 指针所指向的 Snapshot但不会影响前面的 Snapshot 的读操作。
当一切准备完毕之后会以原子操作的方式 commit 这个 Metadata 文件这样一次 Iceberg 的数据写入就完成了。随着每次的写入Iceberg 就生成了下图这样的一个文件组织模式。
2.1.2 增量读取数据
Iceberg 的每个 Snapshot 都包含前一个 Snapshot 的所有数据每次都相当于全量读取数据对于整个链路来说读取数据的代价是非常高的。
如果我们只想读取当前时刻的增量数据就可以根据 Iceberg 中 Snapshot 的回溯机制来实现仅读取 Snapshot-1 到 Snapshot-2 的增量数据也就是下图中的紫色数据部分。
同理S3 也可以只读取红色部分的增量数据也可以读取 S1 - S3 的增量数据。
Iceberg 支持读写分离也就是说可以支持并发读和增量读。 2.1.3 原子性操作
对于文件列表的所有修改都是原子操作。
在分区中追加数据。合并或是重写分区。 Iceberg 是以 文件 为粒度提交事务的所以就没有办法做到以秒为单位提交事务否则会造成文件数据量膨胀。比如 Flink 是以 CheckPoint 为写入单位物理数据在写入 Iceberg 之后并不能被直接查询只有当触发了 CheckPoint 时才会写 Metadata这时数据才会由不可见变成可见。而每次 CheckPoint 执行也需要一定的时间。
2.2 事务性提交
2.2.1 写操作要求
原子性替换保证了线性的历史。原子性替换需要依靠以下操作来保证
记录当前元数据的版本 base version。创建新的元数据以及 Manifest 文件。原子性的将 base version 替换为新的版本。
2.2.2 冲突解决 - 乐观锁
假定当前没有其他的写操作。遇到冲突则基于当前最新的元数据进行重试。元数据管理器所提供的能力。HDFS 或是本地文件系统所提供的原子化的 rename 能力。
3.Iceberg 结合 Flink 场景分享
3.1 构建近实时 Data Pipeline
Iceberg 可以做到分钟级别的准实时数据拉取。 首先Flink Iceberg 最经典的一个场景就是 构建实时的 Data Pipeline。业务端产生的大量日志数据被导入到 Kafka 这样的消息队列。运用 Flink 流计算引擎执行 ETL 后导入到 Apache Iceberg 原始表中。有一些业务场景需要直接跑分析作业来分析原始表的数据而另外一些业务需要对数据做进一步的提纯。那么我们可以再新起一个 Flink 作业从 Apache Iceberg 表中消费增量数据经过处理之后写入到提纯之后的 Iceberg 表中。此时可能还有业务需要对数据做进一步的聚合那么我们继续在 Iceberg 表上启动增量 Flink 作业将聚合之后的数据结果写入到聚合表中。
有人会想这个场景好像通过 Flink Hive 也能实现。 Flink Hive 的确可以实现但写入到 Hive 的数据更多地是为了实现数仓的数据分析而不是为了做增量拉取。一般来说Hive 的增量写入以 Partition 为单位时间是 15 m i n 15min 15min 以上Flink 长期高频率地写入会造成 Partition 膨胀。而 Iceberg 容许实现 1 m i n 1min 1min 甚至 30 s 30s 30s 的增量写入这样就可以大大提高了端到端数据的实时性上层的分析作业可以看到更新的数据下游的增量作业可以读取到更新的数据。
3.2 CDC 数据实时摄入摄出
Flink CDCChange Data Capture增量数据写入 Iceberg。
支持准实时的数据入湖和数据分析。计算引擎原生支持 CDC无需添加额外的组件。采用统一的数据湖存储方案并支持多种数据分析引擎。支持增量数据读取。 可以用 Flink Iceberg 来分析来自 MySQL 等关系型数据库的 binlog 等。一方面Apache Flink 已经原生地支持 CDC 数据解析一条 binlog 数据通过 ververica flink-cdc-connector 拉取之后自动转换成 Flink Runtime 能识别的 INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER 四种消息供用户做进一步的实时计算。
此外CDC 数据成功入湖 Iceberg 之后我们还会打通常见的计算引擎例如 Presto、Spark、Hive 等他们都可以实时地读取到 Iceberg 表中的最新数据。 MySQL Binlog 是二进制格式的日志文件但是不能把 binlog 文件等同于 OS 系统某目录下的具体文件这是狭隘的。Binlog 是用来记录 MySQL 内部对数据库的改动只记录对数据的修改操作主要用于数据库的主从复制以及增量恢复。 3.3 从 Iceberg 历史数据启动 Flink 任务 上面的架构是采用 Iceberg 全量数据和 Kafka 的增量数据来驱动新的 Flink 作业。如果需要过去很长时间例如一年的数据可以采用常见的 Lambda 架构离线链路通过 Kafka → Flink → Iceberg 同步写入到数据湖由于 Kafka 成本较高保留最近 7 7 7 天数据即可Iceberg 存储成本较低可以存储全量的历史数据启动新 Flink 作业的时候只需要去拉 Iceberg 的数据跑完之后平滑地对接到 Kafka 数据即可。
3.4 通过 Iceberg 数据来修正实时聚合结果 同样是在 Lambda 架构下实时链路由于事件丢失或者到达顺序的问题可能导致流计算端结果不一定完全准确这时候一般都需要全量的历史数据来订正实时计算的结果。而我们的 Iceberg 可以很好地充当这个角色因为它可以高性价比地管理好历史数据。
4.Iceberg 0.11.1 源代码编译
4.1 编译 Iceberg
构建 Iceberg 需要 Grade 5.6 5.6 5.6 和 Java 8 8 8 的环境。
4.1.1 下载 Iceberg 0.11.1 软件包
下载地址
https://github.com/apache/iceberg/releases/tag/apache-iceberg-0.11.1https://www.apache.org/dyn/closer.cgi/iceberg/apache-iceberg-0.11.0/apache-iceberg-0.11.0.tar.gz
4.1.2 解压 Iceberg 0.11.1 软件包
[bigdatabigdata185 software]$ tar -zxvf iceberg-apache-iceberg-0.11.1.tar.gz -C /opt/module/
[bigdatabigdata185 software]$ cd /opt/module/iceberg-apache-iceberg-0.11.1/4.1.3 修改对应的版本
我们选择最稳定的版本进行编译Hadoop 2.7.7 2.7.7 2.7.7、Hive 2.3.9 2.3.9 2.3.9、Flink 1.11.6 1.11.6 1.11.6、Spark 3.0.3 3.0.3 3.0.3。
org.apache.flink:* 1.11.6
org.apache.hadoop:* 2.7.7
org.apache.hive:hive-metastore 2.3.9
org.apache.hive:hive-serde 2.3.9
org.apache.spark:spark-hive_2.12 3.0.34.1.4 编辑 build.gradle 文件添加国内源
1在 buildscript 的 repositories 中添加
maven { url https://mirrors.huaweicloud.com/repository/maven/ }添加后如下所示
buildscript {repositories {jcenter()gradlePluginPortal()maven { url https://mirrors.huaweicloud.com/repository/maven/ }}dependencies {classpath com.github.jengelman.gradle.plugins:shadow:5.0.0classpath com.palantir.baseline:gradle-baseline-java:3.36.2classpath com.palantir.gradle.gitversion:gradle-git-version:0.12.3classpath com.diffplug.spotless:spotless-plugin-gradle:3.14.0classpath gradle.plugin.org.inferred:gradle-processors:2.1.0classpath me.champeau.gradle:jmh-gradle-plugin:0.4.8}
}2allprojects 中添加
maven { url https://mirrors.huaweicloud.com/repository/maven/ }添加后如下所示
allprojects {group org.apache.icebergversion getProjectVersion()repositories {maven { url https://mirrors.huaweicloud.com/repository/maven/ }mavenCentral()mavenLocal()}
}4.1.5 下载依赖可选
进入项目根目录执行脚本
[bigdatabigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew dependencies4.1.6 正式编译
1进入项目根目录执行
[bigdatabigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build2上述命令会执行代码里的单元测试如果不需要则执行以下命令
[bigdatabigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build -x test -x scalaStyle4.1.7 生成的目录 4.2 Iceberg 环境部署
在后面的章节中我们分别介绍如何集成 Iceberg 0.11.1 和 Flink 1.11.6、Spark 3.0.3、Hive 2.3.9。
5.总结
数据湖的解决方案 Iceberg 介绍。Apache Iceberg 的技术实现细节。Iceberg 结合 Flink 场景分享。Iceberg 0.11.1 0.11.1 0.11.1 源码编译。
