第一阶段：以Hadoop为代表的离线数据处理组件。Hadoop是以HDFS为核心存储，以MapReduce为基本计算模型的批量数据处理基础组件。围绕HDFS和MR，为不断完善大数据平台的数据处理能力，先后诞生了一系列大数据组件，例如面向实时KV操作的HBase、面向SQL的hive、面向工作流的Pig等。同时，随着大家对于批处理的性能要求越来越高，新的计算模型不断被提出，产生了Tez、Spark、Presto等计算引擎，MR模型也逐渐进化成DAG模型。
为减少数据处理过程中的中间结果写文件操作，Spark、Presto等计算引擎尽量使用计算节点的内存对数据进行缓存，从而提高整个数据过程的效率和系统吞吐能力。
1.2 Lambda架构 随着数据处理能力和处理需求的不断变化，越来越多的用户发现，批处理模式无论如何提升性能，也无法满足实时性要求高的处理场景，流式计算引擎应运而生，例如Storm、Spark Streaming、Flink等。
然而，随着越来越多的应用上线，大家发现，其实批处理和流计算配合使用，才能满足大部分应用需求，对实时性要求高的场景，就会使用Flink Kafka的方式构建实时流处理平台，来满足用户的实时需求。于是Lambda架构被提出，如下图所示。
Lambda架构的核心理念是“流批分离”，如上图所示，整个数据流向自左向右流入平台。进入平台后一分为二，一部分走批处理模式，一部分走流式计算模式。无论哪种计算模式，最终的处理结果都通过服务层对应用提供，确保访问的一致性。
这种数据架构包含非常多的大数据组件，很大程度上增强了整体架构的复杂性和维护成本。
1.3 Lambda架构的痛点 经过多年的发展，Lambda架构比较稳定，能满足过去的应用场景。但是它有很多致命的弱点：
1、数据治理成本高
实时计算流程无法复用离线数仓的数据血缘、数据质量管理体系。需要重新实现一套针对实时计算的数据血缘、数据质量管理体系。
2、开发维护成本高
需要同时维护离线和实时两套数据仓库系统，同一套计算逻辑要存储两份数据。例如，某一条或几条原式数据的更新，就需要重新跑一遍离线数据仓库，数据更新成本非常大。
3、数据口径不一致
因为离线和实时计算走的是两个完全不同的代码，由于数据数据的延迟到达和两类代码运行的时间不一样，导致计算结果不一致。
那么有没有一种架构能解决Lambda架构的问题呢？
1.4 Kappa架构 Lambda架构的“流批分离”处理链路增大了研发的复杂性。因此，有人就提出能不能用一套系统来解决所有问题。目前比较流行的做法就是基于流计算来做。接下来我们介绍一下Kappa架构，通过Flink Kafka将整个链路串联起来。Kappa架构解决了Lambda架构中离线处理层和实时处理层之间计算引擎不一致，开发、运维成本成本高，计算结果不一致等问题。
Kappa架构的方案也被称为“批流一体化”方案。我们借用Flink Kafka来构建流批一体化场景，当需要对ODS层数据做进一步的分析时，将Flink计算结果的DWD层写入到Kafka，同样也会将一部分DWS层的计算结果Kafka。Kappa架构也不是完美的，它也有很多痛点。
1.5 Kappa架构的痛点
1、数据回溯能力弱
但是Kafka对复杂的需求分析支持能力弱，在面对更复杂的数据分析时，又要将DWD和DWS层的数据写入到ClickHouse、ES、MySQL或者是Hive里做进一步分析，这无疑带来了链路的复杂性。更大的问题是在做数据回溯时，由于链路的复杂性导致数据回溯能力非常弱。
2、OLAP分析能力弱
由于Kafka是一个顺序存储的系统，顺序存储系统是没有办法直接在其上进行OLAP分析的，例如谓词下推这类的优化策略，在顺序存储平台（Kafka）上实现是比较困难的事情。
3、数据时序性受到挑战
Kappa架构是严重依赖于消息队列的，我们知道消息队列本身的准确性严格依赖它上游数据的顺序，但是，消息队列的数据分层越多，发生乱序的可能性越大。通常情况下，ODS层的数据是绝对准确的，把ODS层数据经过计算之后写入到DWD层时就会产生乱序，DWD到DWS更容易产生乱序，这样的数据不一致性问题非常大。
1.6 大数据架构痛点总结 从传统的hadoop架构往Lambda架构，从Lambda架构往Kappa架构的演进，大数据平台基础架构的演进逐渐囊括了应用所需的各类数据处理能力，但是这些平台仍然存在很多痛点。
是否存在一种存储技术，既能够支持数据高效的回溯能力，支持数据的更新，又能够实现数据的批流读写，并且还能够实现分钟级到秒级的数据接入？
1.7 实时数仓建设需求 这也是实时数仓建设的迫切需求。实际上是可以通过对Kappa架构进行升级，以解决Kappa架构中遇到的一些问题，接下来主要分享当前比较火的数据湖技术。
那么有没有这样一个架构，既能够满足实时性的需求，又能够满足离线计算的要求，而且还能够减轻开发运维的成本，解决通过消息队列方式构建的Kappa架构中遇到的痛点？答案是肯定的，在文章的后面会详细论述。
二 数据湖助力于解决数据仓库痛点问题2.1 不断完善的数据湖理念 数据湖是一个集中式存储库，可以存储结构化和非结构化数据。可以按业务数据的原样存储（无需先对数据进行结构化处理），并运行不同类型的分析 – 从控制面板和可视化到大数据处理、实时分析和机器学习，以指导做出更好的决策。
2.1.1 存储原式数据
2.2.1 分布式文件系统 第一层是分布式文件系统，对于选择云上技术的用户，通常会选择S3和阿里云存储数据；喜欢开源技术的用户一般采用自己维护的HDFS存储数据。
2.2.2 数据加速层 第二层是数据加速层。数据湖架构是一个典型的存储计算分离架构，远程读写的性能损耗非常大。我们常见的做法是，把经常访问的数据（热点数据）缓存在计算节点本地，从而实现数据的冷热分离。这样做的好处是，提高数据的读写性能，节省网络带宽。我们可以选择开源的Alluxio，或者阿里云的Jindofs。
2.2.3 Table format 层 第三层是Table format层，把数据文件封装成具有业务含义的表，数据本身提供ACID、Snapshot、schema、partition等表级别的语义。这一层可以选择开源数据湖三剑客Delta、Iceberg、Hudi之一。Delta、Iceberg、Hudi是构建数据湖的一种技术，它们本身并不是数据湖。
2.2.4 计算引擎 第四层是各种数据计算引擎。包括Spark、Flink、Hive、Presto等，这些计算引擎都可以访问数据湖中的同一张表。
三 数据湖和数据仓库理念的对比3.1 数据湖和数据仓库对比 下面跟大家聊聊我所理解的数据湖的本质，对于一种新事物不了解本质，你就很难驾驭它，下面这张图道尽了一切。
对数据湖的概念有了基本的认知之后，我们需要进一步明确数据湖需要具备哪些基本特征，特别是与数据仓库相比，数据湖具有哪些特点。我们引用一下AWS数据仓库和数据湖对比官方对比表格。
每个公司需要数据仓库和数据湖，因为它们分别满足不同的需要和使用案例：
上表介绍了数据湖与传统数据仓库的区别，下面我们将从数据存储和计算两个层面进一步分析数据湖应该具备哪些特征。
3.2 写时模式和读时模式3.2.1 写时模式 数据仓库的“写入型Schema”背后隐藏的逻辑就是在数据写入之前，必须确认好数据的Schema，然后进行数据导入，这样做的好处是：可以把业务和数据很好的结合在一起；不足就是在业务模式不清晰，还处于探索阶段时，数仓的灵活性不够。
3.2.2 读时模式 数据湖强调的是“读取型Schema”，背后潜在的逻辑是，认为业务的不确定性是常态：既然我们无法预测业务的发展变化，那么我们就保持一定的灵活性。将结构化设计延后，让整个基础设施具备使数据“按需”贴合业务的能力。因此，数据湖更适合发展、创新型企业。
3.3 数据仓库开发流程
数据湖采用的是灵活，快速的“读时模式” ，在数字化转型的浪潮下真正帮助企业完成技术转型，完成数据沉淀，应对企业快速发展下层出不穷的数据需求问题。
3.4 数据湖的架构方案 数据湖可以认为是新一代的大数据基础设施。在这套架构中，无论是数据的流式处理，还是批处理，数据存储都统一到数据湖-Iceberg上。很明显，这套架构可以解决Lambda架构和Kappa架构的痛点问题：
3.4.1 解决Kafka存储数据量少的问题 目前所有数据湖基本思路都是基于HDFS之上实现的一个文件管理系统，所以数据体量可以很大。
3.4.2 支持OLAP查询 同样数据湖基于HDFS之上实现，只需要当前的OLAP查询引擎做一些适配，就可以对中间层数据进行OLAP查询。
3.4.3 数据治理一体化 批流的数据在HDFS、S3等介质上存储之后，就完全可以复用一套相同的数据血缘、数据质量管理体系。
3.4.4 流批架构统一 数据湖架构相比Lambad架构来说，schema统一，数据处理逻辑统一，用户不再需要维护两份数据。
3.4.5 数据统计口径一致 由于采用统一的流批一体化计算和存储方案，因此数据一致性得到了保证。
3.5 孰优孰劣 数据湖和数据仓库，不能说谁更好谁更差，大家都有可取之处，可以实现双方的优势互补，我这里画一张图，方便你的理解：
4.3 实时数仓 - 数据湖分析系统 在建设离线数据仓库时，首先要进行数据接入操作，比如用离线调度系统定时抽取数据，再经过一系列的ETL操作，最后将数据写入到Hive表里面，这个过程的延时比较大。因此，借助于Iceberg的表结构，可以使用Flink，或者Spark Streaming，实现近实时的数据接入，以降低数据延迟性。
基于上面的功能，我们回顾一下前面讨论的Kappa架构，我们已经知道Kappa架构的痛点，Iceberg既然能够作为一个优秀的表格式，又可以支持Streaming reader和Streaming sink。那么，是否可以考虑将Kafka替换成Iceberg？
Iceberg底层依赖的存储是像HDFS或S3这样的廉价存储，并且支持parquet、orc、Avro等存储结构。可以对中间层的结果数据进行OLAP分析。基于Iceberg snapshot的Streaming reader功能，可以把离线任务天级别到小时级别的延迟大大的降低，改造成一个近实时的数据湖分析系统。
在中间处理层，可以用Presto进行一些简单的SQL查询，因为Iceberg支持Streaming Read，所以在系统的中间层也可以直接接入Flink，直接在中间层用Flink做一些批处理或者流式计算的任务，把中间结果做进一步计算后输出到下游。
4.4 Iceberg替换Kafka的劣势
总的来说，Iceberg替换Kafka的优势主要包括：
五 数据湖技术的发展前景 数据湖可能是在下一场大数据技术变革中的亮点，我们需要抓住机遇、抢占先机，一起来学习数据湖。但是我的建议仍然是“学而不用”，为什么这么说呢？例如：在2018年开始的时候，我们一窝蜂的上线Flink，然后一个月一个版本的升级。简直是吃尽了苦头。所以，我们就等互联网大厂把坑填完了，我们再直接短平快的上马数据湖，但是我们一定要学习。
六 总结 通过这篇文章，我们基本了解了什么是数据湖，以及为什么要学习数据湖，它能解决哪些实际问题。后面我们将继续重点讲解为什么要选择Iceberg作为数据湖的解决方案。
第二章 Apache Iceberg概述和源代码的构建
我们在使用不同的引擎进行大数据计算时，需要将数据根据计算引擎进行适配。这是一个相当棘手的问题，为此出现了一种新的解决方案：介于上层计算引擎和底层存储格式之间的一个中间层。这个中间层不是数据存储的方式，只是定义了数据的元数据组织方式，并向计算引擎提供统一的类似传统数据库中"表"的语义。它的底层仍然是Parquet、ORC等存储格式。
基于此，Netflix开发了Iceberg，目前已经是Apache的顶级项目，
在文件Format（parquet/avro/orc等）之上实现Table语义：
上图右侧是Iceberg在数据仓库生态中的位置，和它差不多相当的一个组件是Metastore。不过Metastore是一个服务，而Iceberg就是一系列jar包。对于Table Format，我认为主要包含4个层面的含义，分别是表schema定义（是否支持复杂数据类型），表中文件的组织形式，表相关统计信息、表索引信息以及表的读写API信息。
Iceberg的核心思想，就是在时间轴上跟踪表的所有变化：
从图中可以看到Iceberg将数据进行分层管理，主要分为元数据管理层和数据存储层。元数据管理层又可以细分为三层：
1.5.1 丰富的计算引擎
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后缀结尾，每次更新都产生一个清单列表文件。每行中存储了清单文件的路径。
清单文件里面存储数据文件的分区范围、增加了几个数据文件、删除了几个数据文件等信息。数据文件（Data Files）存储在不同的Manifest Files里面，Manifest Files存储在一个Manifest List文件里面，而一个Manifest List文件代表一个快照。
1.6.1.3 清单文件（Manifest File） [commitUUID]-[attemptID]-[manifestCount].avro：manifest文件
清单文件是以avro格式进行存储的，以avro后缀结尾，每次更新操作都会产生多个清单文件。其里面列出了组成某个快照（snapshot）的数据文件列表。每行都是每个数据文件的详细描述，包括数据文件的状态、文件路径、分区信息、列级别的统计信息（比如每列的最大最小值、空值数等）、文件的大小以及文件里面数据的行数等信息。其中列级别的统计信息在 Scan 的时候可以为算子下推提供数据，以便可以过滤掉不必要的文件。
1.6.2 数据文件 data目录组织形式类似于hive，都是以分区进行目录组织（图中dt为分区列）
Iceberg的数据文件通常存放在data目录下。一共有三种存储格式（Avro、Orc和Parquet），主要是看您选择哪种存储格式，后缀分别对应avro、orc或者parquet。在一个目录，通常会产生多个数据文件。
二 Apache Iceberg的实现细节2.1 快照设计方式2.1.1 快照隔离 可以支持并发读，例如可以同时读取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的回溯机制来实现，仅读取Snapshot1到Snapshot2的增量数据，也就是下图中的紫色数据部分。
同理，S3也可以只读取红色部分的增量数据，也可以读取S1-S3的增量数据。
Iceberg支持读写分离，也就是说可以支持并发读和增量读。
2.1.3 原子性操作 对于文件列表的所有修改都是原子操作。
首先，Flink Iceberg最经典的一个场景就是构建实时的Data Pipeline。业务端产生的大量日志数据，被导入到Kafka这样的消息队列。运用Flink流计算引擎执行ETL后，导入到Apache Iceberg原始表中。有一些业务场景需要直接跑分析作业来分析原始表的数据，而另外一些业务需要对数据做进一步的提纯。那么我们可以再新起一个Flink作业从Apache Iceberg表中消费增量数据，经过处理之后写入到提纯之后的Iceberg表中。此时，可能还有业务需要对数据做进一步的聚合，那么我们继续在Iceberg表上启动增量Flink作业，将聚合之后的数据结果写入到聚合表中。
有人会想，这个场景好像通过Flink Hive也能实现。 Flink Hive的确可以实现，但写入到Hive的数据更多地是为了实现数仓的数据分析，而不是为了做增量拉取。一般来说，Hive的增量写入以Partition为单位，时间是15min以上，Flink长期高频率地写入会造成Partition膨胀。而Iceberg容许实现1分钟甚至30秒的增量写入，这样就可以大大提高了端到端数据的实时性，上层的分析作业可以看到更新的数据，下游的增量作业可以读取到更新的数据。
3.2 CDC数据实时摄入摄出 FlinkCDC增量数据写入Iceberg。
可以用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表中的最新数据。
3.3 从Iceberg历史数据启动Flink任务
上面的架构是采用Iceberg全量数据和Kafka的增量数据来驱动新的Flink作业。如果需要过去很长时间例如一年的数据，可以采用常见的Lambda架构，离线链路通过kafka->flink->iceberg同步写入到数据湖，由于Kafka成本较高，保留最近7天数据即可，Iceberg存储成本较低，可以存储全量的历史数据，启动新Flink作业的时候，只需要去拉Iceberg的数据，跑完之后平滑地对接到kafka数据即可。
3.4 通过Iceberg数据来修正实时聚合结果
同样是在Lambda架构下，实时链路由于事件丢失或者到达顺序的问题，可能导致流计算端结果不一定完全准确，这时候一般都需要全量的历史数据来订正实时计算的结果。而我们的Iceberg可以很好地充当这个角色，因为它可以高性价比地管理好历史数据。
四 Iceberg0.11.1源代码编译4.1 编译Iceberg 构建Iceberg需要Grade5.6和Java8的环境。
4.1.1 下载Iceberg0.11.1软件包 下载地址：
[bigdata@bigdata185 software]$ tar -zxvf iceberg-apache-iceberg-0.11.1.tar.gz -C /opt/module/[bigdata@bigdata185 software]$ cd /opt/module/iceberg-apache-iceberg-0.11.1/4.1.3 修改对应的版本 我们选择最稳定的版本进行编译，Hadoop2.7.7 Hive2.3.9 Flink1.11.6 Spark3.0.3
org.apache.flink:* = 1.11.6org.apache.hadoop:* = 2.7.7org.apache.hive:hive-metastore = 2.3.9org.apache.hive:hive-serde = 2.3.9org.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.0' classpath 'com.palantir.baseline:gradle-baseline-java:3.36.2' classpath 'com.palantir.gradle.gitversion:gradle-git-version:0.12.3' classpath 'com.diffplug.spotless:spotless-plugin-gradle:3.14.0' classpath 'gradle.plugin.org.inferred:gradle-processors:2.1.0' classpath 'me.champeau.gradle:jmh-gradle-plugin:0.4.8' }} （2）allprojects中添加
maven { url 'https://mirrors.huaweicloud.com/repository/maven/'; } 添加后如下所示
allprojects { group = "org.apache.iceberg" version = getProjectVersion() repositories { maven { url 'https://mirrors.huaweicloud.com/repository/maven/'; } mavenCentral() mavenLocal() }}4.1.5 下载依赖（可选） 进入项目根目录，执行脚本：
[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew dependencies
4.1.6 正式编译 （1）进入项目根目录，执行：
[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build （2）上述命令会执行代码里的单元测试，如果不需要，则执行以下命令：
[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ ./gradlew build -x test -x scalaStyle 
4.1.7 生成的目录
4.2 Iceberg环境部署 在后面的章节中，我们分别介绍如何Iceberg0.11.1和Flink1.11.6、Spark3.0.3和Hive2.3.9集成。
五 总结
一 Spark3.0.3 Iceberg0.11.1环境部署1.1 版本声明| Hadoop2.7.7 | Spark3.0.3 | Iceberg0.11.1 |
[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ cp spark3-extensions/build/libs/iceberg-spark3-extensions-0.11.1.jar /opt/module/spark-3.0.3-bin-hadoop2.7/jars/[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ cp spark3-runtime/build/libs/iceberg-spark3-runtime-0.11.1.jar /opt/module/spark-3.0.3-bin-hadoop2.7/jars/
1.3 同步配置文件 将core-site.xml、hdfs-site.xml、hive-site.xml拷贝到Spark安装目录conf下面
cp /opt/module/hadoop-2.7.7/etc/hadoop/core-site.xml /opt/module/spark-3.0.3-bin-hadoop2.7/conf/cp /opt/module/hadoop-2.7.7/etc/hadoop/hdfs-site.xml /opt/module/spark-3.0.3-bin-hadoop2.7/conf/cp /opt/module/hive-2.3.9/conf/hive-site.xml /opt/module/spark-3.0.3-bin-hadoop2.7/conf/
1.4 配置spark参数 配置SparkSQL Catalog，可以使用基于Hive和hadoop两种方式，先演示如何使用hadoop模式。spark.sql.catalog.hive_catalog = org.apache.iceberg.spark.SparkCatalogspark.sql.catalog.hive_catalog.type = hivespark.sql.catalog.hive_catalog.uri = thrift://bigdata185:90831.4.2 Hadoop模式 spark.sql.catalog.hadoop_catalog = org.apache.iceberg.spark.SparkCatalogspark.sql.catalog.hadoop_catalog.type = hadoopspark.sql.catalog.hadoop_catalog.warehouse = hdfs://bigdata185:9000/dw/icebergspark.sql.catalog.catalog-name.type = hadoopspark.sql.catalog.catalog-name.default-namespace = dbspark.sql.catalog.catalog-name.uri = thrift://bigdata185:9083spark.sql.catalog.catalog-name.warehouse= hdfs://bigdata185:9000/dw/iceberg1.4.3 编辑spark-defaults.conf 配置spark参数，配置SparkSQL Catalog，可以用两种方式，基于Hive和基于Hadoop，这里优先选择hadoop；在不特定指定use hadoop_catalog时，默认采用hadoop_catalog。
[bigdata@bigdata185 spark-3.0.3-bin-hadoop2.7]$ vi conf/spark-defaults.conf spark.sql.catalog.hive_catalog = org.apache.iceberg.spark.SparkCatalogspark.sql.catalog.hive_catalog.type = hivespark.sql.catalog.hive_catalog.uri = thrift://bigdata185:9083spark.sql.catalog.hadoop_catalog = org.apache.iceberg.spark.SparkCatalogspark.sql.catalog.hadoop_catalog.type = hadoopspark.sql.catalog.hadoop_catalog.warehouse = hdfs://bigdata185:9000/dw/icebergspark.sql.catalog.catalog-name.type = hadoopspark.sql.catalog.catalog-name.default-namespace = dbspark.sql.catalog.catalog-name.uri = thrift://bigdata185:9083spark.sql.catalog.catalog-name.warehouse= hdfs://bigdata185:9000/dw/iceberg二 SparkSQL Iceberg案例实操 不支持SQL操作的数据湖是没有灵魂的，下面我们将介绍如何在Iceberg中使用Spark SQL。
此时会创建出一个hadoop_catalog.db的数据库，但是看不见这个database。
[bigdata@bigdata185 spark-3.0.3-bin-hadoop2.7]$ bin/spark-sqlspark-sql> show databases;namespacedefaultTime taken: 0.108 seconds, Fetched 1 row(s)spark-sql> 2.1 DDL操作2.1.1 创建Iceberg表 只能创建Iceberg的内部表，不支持外部表。在删除表的时候，数据文件会自动从HDFS上删除。
spark-sql> use hadoop_catalog;spark-sql> create database db;spark-sql> use db;spark-sql> CREATE TABLE sensordata( sensor_id STRING, ts BIGINT, temperature DOUBLE, dt STRING) USING icebergPARTITIONED BY(dt)TBLPROPERTIES ('read.split.target-size'='134217728', 'read.split.metadata-target-size'='33554432');
此时只有sensordata表的Metadata元数据，由于没有插入数据，因此还没有生成数据文件。
2.1.2 查看表结构 （1）DESC table_name
spark-sql> DESC sensordata; （2）DESC FORMATTED table_name
查询出的信息要比DESC详细的多。
spark-sql> DESC FORMATTED sensordata;
2.1.3 ALTER TABLE （1）取消TBLPROPERTIES属性
ALTER TABLE sensordata UNSET TBLPROPERTIES ('read.split.target-size'); （2）设置TBLPROPERTIES属性
ALTER TABLE sensordata SET TBLPROPERTIES ('read.split.target-size'='134217728'); （3）增加列
ALTER TABLE sensordata ADD COLUMNS ( new_column string comment 'new_column docs'); （4）删除列
ALTER TABLE sensordata DROP COLUMN new_column;2.1.4 CTAS 建表 CREATE TABLE hadoop_catalog.db.sensordata_like USING iceberg AS SELECT * FROM sensordata;2.1.5 DROP 表 DROP TABLE hadoop_catalog.db.sensordata_like;DROP TABLE sensordata;2.2 DML操作2.2.1 插入数据 （1）向sensordata表中插入一条记录
spark-sql> CREATE TABLE sensordata( sensor_id STRING, ts BIGINT, temperature DOUBLE, dt STRING) USING icebergPARTITIONED BY(dt);spark-sql> INSERT INTO sensordata VALUES('sensor_01',1635743301,-12.1,'2021-12-01'); （2）查询数据
spark-sql> SELECT * FROM sensordata;
2.2.2 OVER WRITE操作 覆盖操作和Hive一样，会按分区把原始数据重新刷新。
spark-sql> INSERT OVERWRITE sensordata VALUES('sensor_02',1635743301,23.6,'2021-12-01');spark-sql> SELECT * FROM sensordata;
2.2.3 动态覆盖[bigdata@bigdata185 conf]$ vi spark-defaults.conf spark.sql.sources.partitionOverwriteMode=dynamicCREATE TABLE sensordata_01( sensor_id STRING, ts BIGINT, temperature DOUBLE, dt STRING) USING icebergPARTITIONED BY(dt);spark-sql> INSERT INTO sensordata VALUES('sensor_02',1638421701,-22.2,'2021-12-02');spark-sql> SELECT * FROM sensordata;
spark-sql> INSERT OVERWRITE sensordata_01 SELECT * FROM sensordata;spark-sql> SELECT * FROM sensordata_01;
2.2.4 静态覆盖spark-sql> INSERT OVERWRITE sensordata_01 PARTITION(dt='2021-12-31') SELECT sensor_id,ts,temperature FROM sensordata;spark-sql> SELECT * FROM sensordata_01;
2.2.5 删除数据spark-sql> DELETE FROM sensordata_01 WHERE dt>='2021-12-01';
spark-sql> SELECT * FROM sensordata_01;
2.2.6 更新数据 Spark 3.1添加了对UPDATE更新表中匹配行的查询的支持，但是我们使用的Spark3.0.*还不支持。
spark-sql> UPDATE sensordata SET ts=1631349225,sensor_id='sensor_27' WHERE dt='2021-12-01';
2.3 Iceberg查询操作 配置完Catalog之后，就可以在Iceberg中使用Spark查询。
Iceberg支持使用Spark的DataSourceV2 API来查询元数据。Spark DSv2是一个不断发展的API，在不同的Spark版本中有不同的支持级别：
spark-sql> SELECT * FROM hadoop_catalog.db.sensordata_01;SELECT * FROM hadoop_catalog.db.sensordata_01.history;SELECT * FROM hadoop_catalog.db.sensordata_01.files;SELECT * FROM hadoop_catalog.db.sensordata_01.snapshots;2.3.1 历史快照2.3.1.1 历史表 每张Iceberg表都有一张对应的历史表，历史的表名是当前表加上后缀.history，注意：查询历史快照表的时候必须是表的全称，不能切换到hadoop_prod.db库，再查询历史表。历史表数据表示该表被操作过几次。
SELECT * FROM hadoop_catalog.db.sensordata_01.history;SELECT * FROM hadoop_catalog.db.sensordata.history;
2.3.1.2 快照表 在了解操作次数之后，可以进一步查看每次操作对应的快照记录，对应的快照表是在原始表的基础上追加.snapshots，同样必须是表的全称而不能是简写。
SELECT * FROM hadoop_catalog.db.sensordata_01.snapshots;
可以看见commit时间，snapshot的快照id，parent_id父节点，operation操作类型，summary概要，summary概要字段中包括数据量大小，总条数等记录信息。
SELECT h.made_current_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY made_current_at;
2.3.1.3 查询历史快照信息 知道了快照表和历史表的信息后，可以根据快照id查询数据表在某个历史时期的数据情况，以检测是否有操作错误。如果是误操作，则可以通过spark重新刷新数据。查询方式如下：
scala> :pastespark.read.option("snapshot-id","8109378507001106084").format("iceberg").load("/dw/iceberg/db/sensordata_01").show
2.3.2 files 可以查询表的数据文件和每个文件的元数据。
SELECT * FROM hadoop_catalog.db.sensordata.files;
2.3.3 Manifests 也可以查询表文件的Manifests，这是Iceberg表的元数据文件，后面会重点介绍这部分内容。
SELECT * FROM hadoop_catalog.db.sensordata_01.manifests;
2.4 自动分区（隐藏分区）2.4.1 days函数 在演示完分区表之后，我们接下来讲解创建隐藏分区表。隐藏分区支持的函数有days，years，months，hours，bucket和truncate。接下来演示创建sensordata_days分区表。
CREATE TABLE sensordata_days( sensor_id STRING, ts TIMESTAMP, temperature DOUBLE) USING icebergPARTITIONED BY(days(ts)); 向表中插入不同日期的数据。
INSERT INTO sensordata_days VALUES('sensor_01',CAST(1639816497 AS TIMESTAMP),10.2),('sensor_01',CAST(1639730097 AS TIMESTAMP),12.3); 插入成功之后，再查询表的数据。
spark-sql> SELECT * FROM sensordata_days;
可以看到有两条数据，并且日期不是同一天，再查看HDFS上对应的分区，已经自动按天进行了分区。
2.4.2 years函数 接下来演示创建sensordata_years分区表。
CREATE TABLE sensordata_years( sensor_id STRING, ts TIMESTAMP, temperature DOUBLE) USING icebergPARTITIONED BY(years(ts)); 同样，向表中插入不同年份的数据。
INSERT INTO sensordata_years VALUES('sensor_01',CAST(1608194097 AS TIMESTAMP),10.2),('sensor_01',CAST(1639730097 AS TIMESTAMP),12.3); 插入成功之后，再查询表的数据。
SELECT * FROM sensordata_years;
再查看对应的HDFS地址，已经按照年份创建好分区。
2.4.3 months函数 接下来演示创建sensordata_months分区表。
CREATE TABLE sensordata_months( sensor_id STRING, ts TIMESTAMP, temperature DOUBLE) USING icebergPARTITIONED BY(months(ts)); 同样，向表中插入不同月份的数据。
INSERT INTO sensordata_months VALUES('sensor_01',CAST(1637138097 AS TIMESTAMP),10.2),('sensor_01',CAST(1639730097 AS TIMESTAMP),12.3); 插入成功之后，再查询表的数据。
spark-sql> SELECT * FROM sensordata_months;
再查看对应的HDFS地址，已经按照月份创建好分区。
三 Iceberg表格式详述 Apache Iceberg作为一款新兴的数据湖解决方案，在存储上能够对接当前主流的HDFS，S3文件系统,并且支持多种文件存储格式，例如Parquet、ORC、AVRO。可以与多种计算引擎对接，目前社区已经支持Spark、Flink读写Iceberg、Presto、Hive查询Iceberg。
3.1 Iceberg表格架构图 为方便大家学习Iceberg主要解决Hive的哪些不足，我们先看一下Iceberg表格式的架构图。
Iceberg表结构的架构一共分成3层：
3.3 Iceberg Catalog
当我们想操作Iceberg表数据时，必须先定位出从哪里去哪里读/写该表的数据；也就是说，第一步是找到当前元数据的指针位置。
而Iceberg Catalog里面就存放着指针的相关信息。并且更新当前元数据指针的操作都是原子操作（HDFS、Hive Metastore），这Iceberg表提供事务性操作保证的基础。
在Iceberg Catalog中，每张表都有一个指向该表元数据文件的指针，例如在使用HDFS作为Catalog时，version-hint.text是表元数据的指针，里面的内容是当前元数据文件的版本号。
因此，当SELECT查询Iceberg表时，查询引擎首先进入Iceberg目录，然后检索它要读取的表的当前元数据文件的位置条目，然后打开该文件。3.4 元数据层 而元数据管理层又可以细分为三层：
下面以HadoopTableOperation commit生成的数据为例，介绍各层的数据格式。iceberg生成的数据目录结构如下所示：
.└── sensordata ├── data │ ├── dt=2021-12-01 │ │ ├── 00000-0-275a936f-4d21-4a82-9346-bceac4381e6c-00001.parquet │ │ └── 00000-2-1189ac19-b488-4956-8de8-8dd96cd5920a-00001.parquet │ └── dt=2021-12-02 │ └── 00000-1-cc4f552a-28eb-4ff3-a4fa-6c28ce6e5f79-00001.parquet └── metadata ├── 0dafa6f3-2dbd-4728-ba9b-af31a3416700-m0.avro ├── 2b1fbd5a-6241-4f7d-a4a6-3121019b9afb-m0.avro ├── 2b1fbd5a-6241-4f7d-a4a6-3121019b9afb-m1.avro ├── ad4cd65e-7351-4ad3-baaf-5e5bd99dc257-m0.avro ├── snap-232980156660427676-1-0dafa6f3-2dbd-4728-ba9b-af31a3416700.avro ├── snap-4599216928086762873-1-ad4cd65e-7351-4ad3-baaf-5e5bd99dc257.avro ├── snap-5874199297651146296-1-2b1fbd5a-6241-4f7d-a4a6-3121019b9afb.avro ├── v1.metadata.json ├── v2.metadata.json ├── v3.metadata.json ├── v4.metadata.json └── version-hint.text 其中metadata目录存放元数据管理层的数据：
SELECT h.made_current_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY made_current_at; 下面是一个元数据文件的完整内容，v4.metadata.json
{ // 当前文件格式版本信息，目前为version 1，支持row-level delete等功能的version 2还在开发中 "format-version" : 1, "table-uuid" : "3e0d4750-bf7d-4ace-95a3-881732103f86", // hadoopTable location "location" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata", // 最新snapshot的创建时间 "last-updated-ms" : 1642474805225, "last-column-id" : 4, // iceberg schema "schema" : { "type" : "struct", "fields" : [ { "id" : 1, "name" : "sensor_id", "required" : false, "type" : "string" }, { "id" : 2, "name" : "ts", "required" : false, "type" : "long" }, { "id" : 3, "name" : "temperature", "required" : false, "type" : "double" }, { "id" : 4, "name" : "dt", "required" : false, "type" : "string" } ] }, "partition-spec" : [ { "name" : "dt", "transform" : "identity", "source-id" : 4, "field-id" : 1000 } ], "default-spec-id" : 0, // 分区信息 "partition-specs" : [ { "spec-id" : 0, "fields" : [ { "name" : "dt", // transform类型：目前支持identity，year，bucket等 "transform" : "identity", // 对应schema.fields中相应field的dt "source-id" : 4, "field-id" : 1000 } ] } ], "default-sort-order-id" : 0, "sort-orders" : [ { "order-id" : 0, "fields" : [ ] } ], // Iceberg表的property信息 "properties" : { "owner" : "bigdata" }, // 当前snapshot id "current-snapshot-id" : 4599216928086762873, // snapshot信息 "snapshots" : [ { "snapshot-id" : 232980156660427676, // 创建snapshot时间 "timestamp-ms" : 1642474364978, "summary" : { // spark写入方式，目前支持overwrite以及append "operation" : "append", "spark.app.id" : "local-1642474286024", // 本次snapshot添加的文件数量 "added-data-files" : "1", // 本次snapshot添加的record数量 "added-records" : "1", // 本次snapshot添加的文件大小 "added-files-size" : "1247", // 本次snapshot修改的分区数量 "changed-partition-count" : "1", // 本次snapshot中record总数 = lastSnapshotTotalRecord - currentSnapshotDeleteRecord currentSnapshotAddRecord "total-records" : "1", "total-data-files" : "1", "total-delete-files" : "0", "total-position-deletes" : "0", "total-equality-deletes" : "0" }, // snapshot文件路径 "manifest-list" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/snap-232980156660427676-1-0dafa6f3-2dbd-4728-ba9b-af31a3416700.avro" }, { "snapshot-id" : 5874199297651146296, // 上次snapshotID "parent-snapshot-id" : 232980156660427676, "timestamp-ms" : 1642474492951, "summary" : { "operation" : "overwrite", "spark.app.id" : "local-1642474286024", "added-data-files" : "1", "deleted-data-files" : "1", "added-records" : "1", "deleted-records" : "1", "added-files-size" : "1247", "removed-files-size" : "1247", "changed-partition-count" : "1", "total-records" : "1", "total-data-files" : "1", "total-delete-files" : "0", "total-position-deletes" : "0", "total-equality-deletes" : "0" }, "manifest-list" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/snap-5874199297651146296-1-2b1fbd5a-6241-4f7d-a4a6-3121019b9afb.avro" }, { "snapshot-id" : 4599216928086762873, "parent-snapshot-id" : 5874199297651146296, "timestamp-ms" : 1642474805225, "summary" : { "operation" : "append", "spark.app.id" : "local-1642474751012", "added-data-files" : "1", "added-records" : "1", "added-files-size" : "1246", "changed-partition-count" : "1", "total-records" : "2", "total-data-files" : "2", "total-delete-files" : "0", "total-position-deletes" : "0", "total-equality-deletes" : "0" }, "manifest-list" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/snap-4599216928086762873-1-ad4cd65e-7351-4ad3-baaf-5e5bd99dc257.avro" } ], // snapshot记录 "snapshot-log" : [ { "timestamp-ms" : 1642474364978, "snapshot-id" : 232980156660427676 }, { "timestamp-ms" : 1642474492951, "snapshot-id" : 5874199297651146296 }, { "timestamp-ms" : 1642474805225, "snapshot-id" : 4599216928086762873 } ], // metada记录 "metadata-log" : [ { "timestamp-ms" : 1642474310215, "metadata-file" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/v1.metadata.json" }, { "timestamp-ms" : 1642474364978, "metadata-file" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/v2.metadata.json" }, { "timestamp-ms" : 1642474492951, "metadata-file" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/v3.metadata.json" } ]} 上面展示的是v4.metadata.json中的数据，该文件保存了Iceberg table schema、partition、snapshot信息，partition中的transform信息使得Iceberg能够根据字段进行hidden partition，而无需像hive一样显示的指定分区字段。由于Metadata中记录了每次snapshot的id以及create_time，我们可以通过时间或snapshotId查询相应snapshot的数据，实现Time Travel。
当SELECT查询Iceberg表，并从目录中的表条目获取其位置后打开其当前元数据文件时，查询引擎先读取current-snapshot-id。然后它使用该值在snapshots数组中查找该快照的条目，然后检索该快照的manifest-list条目的值，并打开该位置指向的清单列表。
3.4.2 清单列表文件（Manifest list） Manifest list也被称为（Snapshot），一个snapshot中可以包含多个manifest entry，一个manifest entry表示一个manifest，其中重点需要关注的是每个manifest中的partitions字段，在根据filter进行过滤时可以首先通过该字段表示的分区范围对manifest进行过滤，避免无效的查询。
下面是一个清单列表文件的完整内容，java -jar /opt/sources/avro-tools-1.11.0.jar tojson snap-4599216928086762873-1-ad4cd65e-7351-4ad3-baaf-5e5bd99dc257.avro 转换后的json文件内容如下
{ "manifest_path": "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/ad4cd65e-7351-4ad3-baaf-5e5bd99dc257-m0.avro", "manifest_length": 5932, "partition_spec_id": 0, // 该manifest entry所属的snapshot "added_snapshot_id": { "long": 4599216928086762873 }, // 该manifest中添加的文件数量 "added_data_files_count": { "int": 1 }, // 创建该manifest时已经存在且没有被这次创建操作删除的文件数量 "existing_data_files_count": { "int": 0 }, // 创建manifest时删除的数据文件数量 "deleted_data_files_count": { "int": 0 }, // 该manifest中partition字段的范围 "partitions": { "array": [{ "contains_null": false, "lower_bound": { "bytes": "2021-12-02" }, "upper_bound": { "bytes": "2021-12-02" } }] }, "added_rows_count": { "long": 1 }, "existing_rows_count": { "long": 0 }, "deleted_rows_count": { "long": 0 }} // manifest entry{ "manifest_path": "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/2b1fbd5a-6241-4f7d-a4a6-3121019b9afb-m1.avro", "manifest_length": 5933, "partition_spec_id": 0, "added_snapshot_id": { "long": 5874199297651146296 }, "added_data_files_count": { "int": 1 }, "existing_data_files_count": { "int": 0 }, "deleted_data_files_count": { "int": 0 }, "partitions": { "array": [{ "contains_null": false, "lower_bound": { "bytes": "2021-12-01" }, "upper_bound": { "bytes": "2021-12-01" } }] }, "added_rows_count": { "long": 1 }, "existing_rows_count": { "long": 0 }, "deleted_rows_count": { "long": 0 }} 当SELECT查询Iceberg表并在从元数据文件中获取快照的位置后，并打开清单列表时，查询引擎会读取清单路径条目的值，并打开清单文件。
3.4.3 清单文件
java -jar /opt/sources/avro-tools-1.11.0.jar tojson 2b1fbd5a-6241-4f7d-a4a6-3121019b9afb-m1.avro 转换后的json文件内容如下，里面的file_path是数据文件的地址：{ // 表示对应数据文件status // 0: EXISTING, 1: ADDED，2: DELETED "status": 1, "snapshot_id": { "long": 5874199297651146296 }, "data_file": { "file_path": "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/data/dt=2021-12-01/00000-2-1189ac19-b488-4956-8de8-8dd96cd5920a-00001.parquet", "file_format": "PARQUET", // 对应的分区值 "partition": { "dt": { "string": "2021-12-01" } }, // 文件中record数量 "record_count": 1, "file_size_in_bytes": 1247, "block_size_in_bytes": 67108864, // 不同column存储大小 "column_sizes": { "array": [{ "key": 1, "value": 56 }, { "key": 2, "value": 49 }, { "key": 3, "value": 48 }, { "key": 4, "value": 57 }] }, // 不同列对应的value数量 "value_counts": { "array": [{ "key": 1, "value": 1 }, { "key": 2, "value": 1 }, { "key": 3, "value": 1 }, { "key": 4, "value": 1 }] }, // 列值为null的数量 "null_value_counts": { "array": [{ "key": 1, "value": 0 }, { "key": 2, "value": 0 }, { "key": 3, "value": 0 }, { "key": 4, "value": 0 }] }, "nan_value_counts": { "array": [{ "key": 3, "value": 0 }] }, // 不同列的范围 "lower_bounds": { "array": [{ "key": 1, "value": "sensor_02" }, { "key": 2, "value": "Evau0000u0000u0000u0000" }, { "key": 3, "value": "7@" }, { "key": 4, "value": "2021-12-01" }] }, "upper_bounds": { "array": [{ "key": 1, "value": "sensor_02" }, { "key": 2, "value": "Evau0000u0000u0000u0000" }, { "key": 3, "value": "7@" }, { "key": 4, "value": "2021-12-01" }] }, "key_metadata": null, "split_offsets": { "array": [4] } }} Manifest管理多个data文件，一条DataFileEntry对应一个data文件，DataFileEntry中记录了所属partition，value bounds等信息，value_counts和null_value_counts可以用于过滤null列，除此之外，可以根据value bounds进行过滤，加速查询。
四 总结
在上一章中，我们学习了Iceberg表的不同组件，以及如何访问Iceberg表的中数据，现在我们更深入地学习在Iceberg表上执行CRUD操作时会发生什么。
二 Iceberg表的CRUD操作2.1 CREATE TABLE逻辑分析2.1.1 创建Iceberg表 我们先删除sensordata表，再重建，以方便分析整个过程，删除表后，数据和元数据都被删除。
spark-sql> use hadoop_catalog.db;spark-sql> show tables;namespace tableNamedb sensordatadb sensordata_hoursdb sensordata_yearsdb sensordata_daysdb sensordata_01db sensordata_monthsTime taken: 1.031 seconds, Fetched 6 row(s)spark-sql> drop table sensordata;spark-sql> CREATE TABLE sensordata( sensor_id STRING, ts BIGINT, temperature DOUBLE, dt STRING) USING icebergPARTITIONED BY(dt);
2.1.2 查看HDFS目录 此时只有sensordata表的Metadata元数据，由于没有插入数据，因此还没有生成数据文件。
2.1.3 建表逻辑分析 我们创建了一个名为sensordata的Iceberg表，一共有4列，并以dt天粒度进行分区。
执行上述操作时，会在元数据层创建一个带有S0快照的元数据文件（快照S0：v1.metadata.json目前不指向任何清单列表，因为表中还不存在数据）。db.sensordata在有数据变更时，将更新当前元数据指针（version-hint.text）的目录条目以指向这个新元数据文件的路径。
2.2 写入数据逻辑分析2.2.1 插入数据 我们先向sensordata表插入两条数据，以方便理解数据插入逻辑。
INSERT INTO sensordata VALUES('sensor_01',1638351932,-0.2,'2021-12-01');2.2.2 查看HDFS目录 此时只有sensordata表开始生成数据文件。
2.2.3 写数据逻辑分析
当我们执行这个INSERT语句时，会发生以下过程：
INSERT INTO sensordata VALUES('sensor_03',1638351932,-23.2,'2021-12-01'),('sensor_02',1638351625,-5.2,'2021-12-02');2.3.1 执行查询操作 SELECT * FROM sensordata;2.3.2 查询操作分析
执行此SELECT语句时，会发生以下过程：
spark-sql> delete from sensordata ;spark-sql> select * from sensordata;spark-sql> SELECT h.made_current_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY made_current_at;
2.4.2 基于快照查询数据 比如我们想查看，删除数据之前的表数据，就可以选择查询snapshot_id=8585349266450764220
scala> :pastespark.read.option("snapshot-id","8585349266450764220").format("iceberg").load("/dw/iceberg/db/sensordata").show2.4.3 查询流程分析
执行此SELECT语句时，会发生以下过程：
可以使用Flink的DataStream API和Table API操作Iceberg表数据，目前仅支持Flink 1.11.x和Iceberg的集成：
为了在Flink中创建Iceberg表，我们推荐从Flink SQL Client入手，这样对于小白来说更容易理解。
1.1 配置参数和jar包1.1.1 配置HADOOP_CLASSPATH 从Flink1.11.0版本开始就不再提供[bigdata@bigdata185 flink-1.11.6]$ vi bin/config.shexport HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`1.1.2 复制Iceberg的jar包 将编译好的[bigdata@bigdata185 iceberg-apache-iceberg-0.11.1]$ cp flink-runtime/build/libs/iceberg-flink-runtime-0.11.1.jar /opt/module/flink-1.11.6/lib/[bigdata@bigdata185 software]$ cp flink-sql-connector-hive-2.3.6_2.12-1.11.6.jar /opt/module/flink-1.11.6/lib/ 备注：
1.1.3 同步配置文件 将core-site.xml、hdfs-site.xml、hive-site.xml拷贝到Flink安装目录conf下面
cp /opt/module/hadoop-2.7.7/etc/hadoop/core-site.xml /opt/module/flink-1.11.6/conf/cp /opt/module/hadoop-2.7.7/etc/hadoop/hdfs-site.xml /opt/module/flink-1.11.6/conf/cp /opt/module/hive-2.3.9/conf/hive-site.xml /opt/module/flink-1.11.6/conf/
1.2 Flink SQL Client1.2.1 启动Flink集群 在hadoop环境中，启动一个单独的Flink集群
[bigdata@bigdata185 ~]$ cd /opt/module/flink-1.11.6/[bigdata@bigdata185 flink-1.11.6]$ bin/start-cluster.sh 1.2.2 启动FlinkSQL Client 相关详细文档请参考：[bigdata@bigdata185 flink-1.11.6]$ bin/sql-client.sh embedded shell 成功启动FlinkSQL Client之后，会出现下面的Flink松鼠图案。
1.3 创建Iceberg的Catalog Flink1.11支持通过Flink sql创建Catalogs。
1.3.1 Hive Catalog 创建一个名为hive_catalog的iceberg catalog ，用来从hive metastore中加载表。
-- 创建hive catalogCREATE CATALOG hive_catalog WITH ( 'type'='iceberg', 'catalog-type'='hive', 'uri'='thrift://bigdata185:9083', 'clients'='5', 'property-version'='1', 'warehouse'='hdfs://bigdata185:9000/dw/iceberg');-- 使用hive catalogUSE CATALOG hive_catalog;
1.3.2 Hadoop Catalog Iceberg还支持HDFS中基于目录的Catalog，使用'catalog-type'='hadoop'进行配置：
-- 创建hadoop catalogCREATE CATALOG hadoop_catalog WITH ( 'type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='hdfs://bigdata185:9000/dw/iceberg', 'property-version'='1');-- 使用hadoop catalogUSE CATALOG hadoop_catalog;
1.3.3 通过YAML创建catalog 在启动SQL客户端之前，Catalogs可以通过在sql-client-defaults.yaml文件中注册。
[bigdata@bigdata185 flink-1.11.6]$ vi conf/sql-client-defaults.yamlcatalogs: - name: hadoop_catalog type: iceberg catalog-type: hadoop warehouse: hdfs://bigdata185:9000/dw/iceberg/
使用hadoop catalog
Flink SQL> use catalog hadoop_catalog; hadoop_catalog创建成功之后，会在HDFS上生成一个目录：
1.3.4 查看Catalog列表 哈哈哈，查看Catalog列表太简单了，敲个show catalogs试试~~~~
Flink SQL> SHOW CATALOGS;
1.3.5 Catalog功能描述 通过上图我们可以知道，使用Flink1.11.6可以创建Catalog。而Catalog是对Iceberg表进行Create、Drop、Rename操作的一个组件。目前Iceberg主要支持HadoopCatalog和HiveCatalog。其中HiveCatalog将当前metadata文件存储在Hive的Metastore中，每次读写Iceberg表都需要从Hive的Metastore中解析数据；而HadoopCatalog将当前表的metadata文件存储在Hadoop的一个路径上面，因此不需要连接Hive Metastore。
二 FlinkSQL操作2.1 创建数据库 （1）再次启动FlinkSQL客户端
[bigdata@bigdata185 flink-1.11.6]$ bin/sql-client.sh embedded shellFlink SQL> USE CATALOG hadoop_catalog; （2）可以使用默认数据库，也可以创建数据库。Iceberg默认会使用Flink中的default数据库。如果我们不想在default数据库下面创建表，可以按照下面的命令创建一个单独的数据库。
Flink SQL> SHOW DATABASES;defaultFlink SQL> Flink SQL> CREATE DATABASE db;
（3）使用数据库
Flink SQL> use db;2.2 创建Iceberg表 我们直接创建分区表，目前Flink对接Iceberg还不能使用Iceberg的隐藏分区特性。
CREATE TABLE sensordata ( sensorid STRING, ts BIGINT, temperature DOUBLE, dt STRING) PARTITIONED BY (dt);
此时只有sensordata表的Metadata元数据，由于没有插入数据，因此还没有生成数据文件。
2.3 LIKE建表 可以通过CREATE TABLE LIKE来创建一个和另外一张表具有相同结构、分区和表属性的相同的表。
CREATE TABLE sensordata_01 LIKE sensordata;
2.4 使用INSERT INTO插入数据 INSERT INTO sensordata VALUES('sensor_01',1638351932,-0.2,'2021-12-01'),('sensor_02',1638351625,-5.2,'2021-12-01');
2.5 Querying with SQL Flink SQL> SELECT * FROM sensordata;
任务监控 可以通过http://bigdata185:8081默认端口查询standlone模式任务是否成功。
2.6 Writing with SQL 从Flink1.11版本开始，支持通过INSERT INTO和INSERT OVERWRITE向Iceberg表写入数据。
2.6.1 INSERT INTO INSERT INTO sensordata VALUES ('sensor_03', 1638533291, 32.1, '2021-12-02');
2.6.2 INSERT OVERWRITE 在Flink的流式（默认）处理环境中，是不支持INSERT OVERWRITE的。
（1）使用OVERWRITE插入
Flink SQL> INSERT OVERWRITE sensordata SELECT 'sensor_04', 1638619691, 31.7, '2021-12-01'; （2）Flink默认使用流的方式插入数据，这个时候流是不支持INSERT OVERWRITE操作的。
（3）需要修改插入模式，改成批的插入模式，再次使用OVERWRITE插入数据。需要改回流式操作时，将参数设置为：SET execution.type = streaming;
Flink SQL> SET execution.type = batch ;Flink SQL> INSERT OVERWRITE sensordata SELECT 'sensor_04', 1638619691, 31.7, '2021-12-01'; （4）查询结果，已经将结果根据分区进行覆盖操作
Flink SQL> SELECT * FROM sensordata;
三 通过Flink DataStream操作Iceberg HadoopCatalog依赖于HDFS提供的rename原子性语义，而HiveCatalog不依赖于任何文件系统的rename原子性语义支持，因此基于HiveCatalog的表不仅可以支持HDFS，而且可以支持S3、OSS等其他文件系统。但是HadoopCatalog可以认为只支持HDFS表，比较难以迁移到其他文件系统。但是HadoopCatalog写入提交的过程只依赖HDFS，不和Metastore/MySQL交互，而HiveCatalog每次提交都需要和Metastore/MySQL交互，可以认为是强依赖于Metastore，如果Metastore有异常，基于HiveCatalog的Iceberg表的写入和查询会有问题。相反，HadoopCatalog并不依赖于Metastore，即使Metastore有异常，也不影响Iceberg表的写入和查询。
3.1 启动Hadoop集群的NameNode和DataNode [bigdata@bigdata185 ~]$ HDFS-2.7.7 start3.2 引入相关maven配置文件 我们采用的是Flink1.11.6 Hadoop2.7.7 Iceberg0.11.1版本组合。
<!-- 通过参数配置版本信息 --> <properties> <flink.version>1.11.6</flink.version> <hadoop.version>2.7.7</hadoop.version> <iceberg.version>0.11.1</iceberg.version> </properties> <dependencies> <!-- 引入Flink相关依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- 引入Iceberg相关依赖 --> <dependency> <groupId>org.apache.iceberg</groupId> <artifactId>iceberg-core</artifactId> <version>${iceberg.version}</version> </dependency> <dependency> <groupId>org.apache.iceberg</groupId> <artifactId>iceberg-flink</artifactId> <version>${iceberg.version}</version> </dependency> <!-- 引入hadoop客户端相关依赖 --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> </dependency> <!-- 引入日志相关依赖 --> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.6.1</version> <!-- 所有的编译都依照JDK1.8来搞 --> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> <!-- 用于项目的打包插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-assembly-plugin</artifactId> <version>3.0.0</version> <configuration> <archive> <manifest> </manifest> </archive> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>3.3 添加log4j.properties配置文件 log4j.rootLogger=WARN, stdoutlog4j.appender.stdout = org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.Target = System.outlog4j.appender.stdout.layout = org.apache.log4j.PatternLayoutlog4j.appender.logfile.layout.ConversionPattern=%d %p [%c]- %m%n3.4 DataStream读数据 Iceberg现在支持使用Java API流式或者批量读取。
3.4.1 批量读取数据 通过batch的方式读取数据
package com.yunclass.iceberg.flinksql;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.data.RowData;import org.apache.iceberg.flink.TableLoader;import org.apache.iceberg.flink.source.FlinkSource;public class TableOperations { public static void main(String[] args) throws Exception { // 设置执行HDFS操作的用户，防止权限不够 System.setProperty("HADOOP_USER_NAME", "bigdata"); // 1 获取Flink的执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 2 使用TableLoader加载HDFS路径 TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://bigdata185:9000/dw/iceberg/db/sensordata"); // 3 操作方式 batchReadTable(env, tableLoader); // 4 执行程序 env.execute("Flink对Iceberg表的批量操作"); } /** * 1、批量读取Iceberg表数据 */ private static void batchReadTable(StreamExecutionEnvironment env, TableLoader tableLoader) { DataStream<RowData> rowData = FlinkSource.forRowData() .env(env) .tableLoader(tableLoader) .streaming(false) .build(); // 将DataStream<RowData>转换为DataStream<String> DataStream<String> outputData = rowData.map(new MapFunction<RowData, String>() { @Override public String map(RowData rowData) throws Exception { String sensorId = rowData.getString(0).toString(); long ts = rowData.getLong(1); double temperature = rowData.getDouble(2); String dt = rowData.getString(3).toString(); return sensorId "," ts "," temperature "," dt; } }); // 打印结果数据 outputData.print(); }}3.4.2 流式读取数据 SELECT h.made_current_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY made_current_at; （1）通过Streaming的方式读取数据
// 2 使用流的方式读取Iceberg表数据 private static void streamingReadTable(StreamExecutionEnvironment env, TableLoader tableLoader) { DataStream<RowData> streamingData = FlinkSource.forRowData() .env(env) .tableLoader(tableLoader) // 流式读取iceberg表数据 .streaming(true) .build(); // 使用lambda表达式的方式打印输出数据 streamingData.map(item -> item.getString(0) "," item.getLong(1) "," item.getDouble(2) "," item.getString(3) ).print(); }} （2）程序在启动之后，不会立刻停止。
（3）因为是流式处理，这个时候需要手动往表中插入一条数据，验证读取情况。
INSERT INTO sensordata VALUES('sensor_05', 1638706091, 21.5, '2021-12-01');INSERT INTO sensordata VALUES('sensor_06', 1638706091, 21.6, '2021-12-01');INSERT INTO sensordata VALUES('sensor_07', 1638706091, 22.7, '2021-12-02');
（4）可以在控制台，准实时打印出新增的数据。
3.5 DataStream写数据 Iceberg支持从不同的DataStream输入，向HDFS上的Iceberg表写入数据。
/** * 3、自定义数据源，追加写入Iceberg表 * @param env */ private static void appendingTable(StreamExecutionEnvironment env) { // 定义随机变量 Random random = new Random(); DecimalFormat df = new DecimalFormat("0.0"); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd"); // 自定义数据源，生成DataStream<RowData> DataStream<RowData> inputData = env.addSource(new SourceFunction<RowData>() { boolean flag = true; @Override public void run(SourceContext<RowData> ctx) throws Exception { GenericRowData rowData = new GenericRowData(4); while (flag) { // 生成5条数据 for (int i = 0; i < 5; i ) { long ts = System.currentTimeMillis(); String sensorId = "sensor_" random.nextInt(10); double temperature = Double.parseDouble(df.format(random.nextDouble() * 10)); String dt = sdf.format(new Timestamp(ts)); rowData.setField(0, StringData.fromString(sensorId)); rowData.setField(1, ts); rowData.setField(2, temperature); rowData.setField(3, StringData.fromString(dt)); ctx.collect(rowData); } cancel(); } } @Override public void cancel() { flag = false; } }); inputData.print(); // 加载目标表 TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://bigdata185:9000/dw/iceberg/db/sensordata_01"); // 向目标表写入数据 FlinkSink.forRowData(inputData) .tableLoader(tableLoader) .overwrite(true) .build(); }3.5.1.2 INSERT INTO SELECT 方式 （1）下面的示例是读取sensordata表的数据，插入到sensordata_01表中，类似于INSERT INTO target_table SELECT * FROM source_table表，执行两遍代码。
// 3 向sensordata_01表appending数据 private static void appendingTable(StreamExecutionEnvironment env, TableLoader tableLoader) { DataStream<RowData> batchData = FlinkSource.forRowData() .env(env) .streaming(false) .tableLoader(tableLoader) .build(); // 加载目标表 TableLoader sinkTable = TableLoader.fromHadoopTable("hdfs://bigdata185:9000/dw/iceberg/db/sensordata_01"); // 向sensordata_01表appending写入数据 FlinkSink.forRowData(batchData) .tableLoader(sinkTable) .build(); System.out.println("ok"); }
（2）采用的是Batch处理，执行两遍代码，查看结果数据。
Flink SQL> select * from sensordata_01;3.5.2 Overwrite data 覆盖写数据3.5.2.1 自定义数据源 编写代码，将overwrite设置为true，sensordata_01的数据将会按照分区自动进行覆盖写操作。
FlinkSink.forRowData(inputData) .tableLoader(tableLoader) .build();3.5.2.2 INSERT OVERWRITE SELECT方式 （1）编写代码，将overwrite设置为true，sensordata_01的数据将会按照分区自动进行覆盖写操作。
// 4、覆盖写数据 public static void overWriteData(StreamExecutionEnvironment env, TableLoader tableLoader) { DataStream<RowData> batchData = FlinkSource.forRowData().env(env).tableLoader(tableLoader).streaming(false).build(); TableLoader tableSensordata = TableLoader.fromHadoopTable("hdfs://bigdata185:8020/flink/warehouse/iceberg_db/sensordata_02"); FlinkSink.forRowData(batchData).tableLoader(tableSensordata).overwrite(true).build(); } （2）查询sensordata_01表查看OverWrite效果，根据分区将数据进行了覆盖操作。
Flink SQL> select * from sensordata_01;
四 操作流程分析 我们先了解一下Iceberg在文件系统中的布局，总体来讲Iceberg分为两部分数据，第一部分是数据文件，如下图中的Parquet文件。第二部分是表元数据文件（Metadata文件），包含Snapshot文件（snap-*.avro）、Manifest 文件(*.avro)、TableMetadata 文件(*.json)等。
4.1 建表逻辑分析4.1.1 创建Iceberg表 我们先删除sensordata表，再重建，以方便分析整个过程，删除表后，数据和元数据都被删除。
Flink SQL> DROP TABLE sensordata;Flink SQL> CREATE TABLE sensordata ( sensorid STRING, ts BIGINT, temperature DOUBLE, dt STRING) PARTITIONED BY (dt);4.1.2 查看HDFS目录 HDFS上只有元数据目录，并没有数据目录
4.1.3 建表逻辑分析 我们创建了一个名为sensordata的Iceberg表，一共有4列，并以dt天粒度进行分区。
在执行建表语句时，会在元数据层创建一个带有S0快照的元数据文件（快照S0目前不指向任何清单列表，因为表中还不存在数据）。db.sensordata在有数据变更时，将更新当前元数据指针的目录条目以指向这个新元数据文件的路径。
4.1.3.1 version-hint.text 在使用HDFS作为Catalog（目录）时，目录中会存在一个version-hint.text文件，文件里面的内容是当前元数据文件的版本号。此时，里面的数字为1，指向v1.metadata.json元数据；这两个数字是一一对应的。
hadoop fs -cat /dw/iceberg/db/sensordata/metadata/version-hint.text 14.1.3.2 metadata 顾名思义，元数据文件存储有关表的元数据。这包括有关表的结构、分区信息、快照以及哪个快照是当前快照的信息。快照S0（current-snapshot-id）目前不指向任何清单列表，因为表中还不存在数据。
元数据文件（v1.metadata.json）样例：
{ "format-version" : 1, "table-uuid" : "7cd325fb-c9af-45f5-ba24-ea7c62624878", "location" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata", "last-updated-ms" : 1641807202347, "last-column-id" : 4, "schema" : { "type" : "struct", "fields" : [ { "id" : 1, "name" : "sensorid", "required" : false, "type" : "string" }, { "id" : 2, "name" : "ts", "required" : false, "type" : "long" }, { "id" : 3, "name" : "temperature", "required" : false, "type" : "double" }, { "id" : 4, "name" : "dt", "required" : false, "type" : "string" } ] }, "partition-spec" : [ { "name" : "dt", "transform" : "identity", "source-id" : 4, "field-id" : 1000 } ], "default-spec-id" : 0, "partition-specs" : [ { "spec-id" : 0, "fields" : [ { "name" : "dt", "transform" : "identity", "source-id" : 4, "field-id" : 1000 } ] } ], "default-sort-order-id" : 0, "sort-orders" : [ { "order-id" : 0, "fields" : [ ] } ], "properties" : { }, "current-snapshot-id" : -1, "snapshots" : [ ], "snapshot-log" : [ ], "metadata-log" : [ ]}4.2 写入数据逻辑分析4.2.1 插入数据 我们先向sensordata表插入两条数据，以方便理解数据插入逻辑。
INSERT INTO sensordata VALUES('sensor_01',1638351932,-0.2,'2021-12-01'),('sensor_02',1638351625,-5.2,'2021-12-01');4.2.2 写入数据逻辑分析 当我们执行这个INSERT语句时，会发生以下过程：
java -jar /opt/sources/avro-tools-1.11.0.jar tojson 1b5f0cc0-a34d-4e9e-ad81-3f3e5bd7bcd3-m0.avro 转换后的json文件内容如下，里面的file_path是数据文件的地址：{ "status": 1, "snapshot_id": { "long": 4271853720390397954 }, "data_file": { "file_path": "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/data/dt=2021-12-01/00000-0-a15bdcbb-f90d-4f89-bbaf-b8ea30187704-00001.parquet", "file_format": "PARQUET", "partition": { "dt": { "string": "2021-12-01" } }, "record_count": 2, "file_size_in_bytes": 1319, "block_size_in_bytes": 67108864, "column_sizes": { "array": [{ "key": 1, "value": 63 }, { "key": 2, "value": 59 }, { "key": 3, "value": 61 }, { "key": 4, "value": 104 }] }, "value_counts": { "array": [{ "key": 1, "value": 2 }, { "key": 2, "value": 2 }, { "key": 3, "value": 2 }, { "key": 4, "value": 2 }] }, "null_value_counts": { "array": [{ "key": 1, "value": 0 }, { "key": 2, "value": 0 }, { "key": 3, "value": 0 }, { "key": 4, "value": 0 }] }, "nan_value_counts": { "array": [{ "key": 3, "value": 0 }] }, "lower_bounds": { "array": [{ "key": 1, "value": "sensor_01" }, { "key": 2, "value": "tC§au0000u0000u0000u0000" }, { "key": 3, "value": "ÍÌÌÌÌÌu0014À" }, { "key": 4, "value": "2021-12-01" }] }, "upper_bounds": { "array": [{ "key": 1, "value": "sensor_02" }, { "key": 2, "value": "<D§au0000u0000u0000u0000" }, { "key": 3, "value": "É¿" }, { "key": 4, "value": "2021-12-01" }] }, "key_metadata": null, "split_offsets": { "array": [4] } }}4.2.2.2 查看清单列表文件（manifest-list） 清单列表文件：java -jar /opt/sources/avro-tools-1.11.0.jar tojson snap-4271853720390397954-1-1b5f0cc0-a34d-4e9e-ad81-3f3e5bd7bcd3.avro 转换后的json文件内容如下，里面的manifest_path是清单文件的地址：{ "manifest_path": "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/1b5f0cc0-a34d-4e9e-ad81-3f3e5bd7bcd3-m0.avro", "manifest_length": 5949, "partition_spec_id": 0, "added_snapshot_id": { "long": 4271853720390397954 }, "added_data_files_count": { "int": 1 }, "existing_data_files_count": { "int": 0 }, "deleted_data_files_count": { "int": 0 }, "partitions": { "array": [{ "contains_null": false, "lower_bound": { "bytes": "2021-12-01" }, "upper_bound": { "bytes": "2021-12-01" } }] }, "added_rows_count": { "long": 2 }, "existing_rows_count": { "long": 0 }, "deleted_rows_count": { "long": 0 }}4.2.2.3 查看元数据文件（v2.metadata.json） 当前的元数据文件内容如下，其中manifest-list指向的是“{ "format-version" : 1, "table-uuid" : "46590bb6-5f0b-4d1b-878f-1b043f1faeb1", "location" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata", "last-updated-ms" : 1641807779230, "last-column-id" : 4, "schema" : { "type" : "struct", "fields" : [ { "id" : 1, "name" : "sensorid", "required" : false, "type" : "string" }, { "id" : 2, "name" : "ts", "required" : false, "type" : "long" }, { "id" : 3, "name" : "temperature", "required" : false, "type" : "double" }, { "id" : 4, "name" : "dt", "required" : false, "type" : "string" } ] }, "partition-spec" : [ { "name" : "dt", "transform" : "identity", "source-id" : 4, "field-id" : 1000 } ], "default-spec-id" : 0, "partition-specs" : [ { "spec-id" : 0, "fields" : [ { "name" : "dt", "transform" : "identity", "source-id" : 4, "field-id" : 1000 } ] } ], "default-sort-order-id" : 0, "sort-orders" : [ { "order-id" : 0, "fields" : [ ] } ], "properties" : { }, "current-snapshot-id" : 4271853720390397954, "snapshots" : [ { "snapshot-id" : 4271853720390397954, "timestamp-ms" : 1641807779230, "summary" : { "operation" : "append", "flink.job-id" : "1cff09f16bf701552a7af2153de0412a", "flink.max-committed-checkpoint-id" : "9223372036854775807", "added-data-files" : "1", "added-records" : "2", "added-files-size" : "1319", "changed-partition-count" : "1", "total-records" : "2", "total-data-files" : "1", "total-delete-files" : "0", "total-position-deletes" : "0", "total-equality-deletes" : "0" }, "manifest-list" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/snap-4271853720390397954-1-1b5f0cc0-a34d-4e9e-ad81-3f3e5bd7bcd3.avro" } ], "snapshot-log" : [ { "timestamp-ms" : 1641807779230, "snapshot-id" : 4271853720390397954 } ], "metadata-log" : [ { "timestamp-ms" : 1641807740051, "metadata-file" : "hdfs://bigdata185:9000/dw/iceberg/db/sensordata/metadata/v1.metadata.json" } ]}4.2.2.4 元数据指针文件（version-hint.text） 次数里面的数字为2，指向v2.metadata.json元数据。
24.3 写入数据流程总结
上方为ceberg数据写入的流程图，这里用计算引擎Flink为例。
1.1 初始化 （1）同步jar包到Hive安装目录下
如果想用Hive操作Iceberg，必须将[bigdata@bigdata185 libs]$ cp iceberg-hive-runtime-0.11.1.jar /opt/module/hive-2.3.9/lib/add jar /opt/jars/iceberg-hive-runtime-0.11.1.jar1.2 Catalog模式1.2.1 Hive Catalog模式 通过Hive Catalog模式管理和操作Iceberg表。
SET iceberg.catalog.hive_catalog.type=hive;SET iceberg.catalog.hive_catalog.uri=thrift://bigdata185:9083;SET iceberg.catalog.hive_catalog.clients=10;SET iceberg.catalog.hive_catalog.warehouse=hdfs://bigdata185:9000/dw/iceberg;1.2.2 Hadoop Catalog模式 通过Hadoop Catalog模式管理和操作Iceberg表。
SET iceberg.catalog.hadoop_catalog.type=hadoop;SET iceberg.catalog.hadoop_catalog.warehouse=hdfs://bigdata185:9000/dw/iceberg;1.3 数据表操作 （1）创建Iceberg表
CREATE DATABASE hive_db;USE hive_db;CREATE TABLE sensordata_hive ( sensor_id STRING, ts BIGINT, temperature DOUBLE) PARTITIONED BY ( dt STRING) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'LOCATION 'hdfs://bigdata185:9000/dw/iceberg/hive_db/sensordata_hive'TBLPROPERTIES ('iceberg.catalog'='hadoop_catalog'); （2）向Iceberg表中写入数据
INSERT INTO sensordata_hive VALUES('sensor_02',1638421701,-22.2,'2021-12-02');
（3）查询数据
SELECT * FROM sensordata_hive;二 Spark Procedures Spark Procedures的功能只有在Spark3.x版本以上才可用，并且要在Catalog中增加Iceberg SQL extensions的支持。
所有的Procedures都存储在system命名表空间中。
vi conf/spark-defaults.conf spark.sql.extensions = org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions2.1 版本管理2.1.1 环境准备 （1）为了在SQL中演示Iceberg的版本管理功能，我们先按照下面的方式准备数据。
-- 创建Iceberg表CREATE TABLE sensordata( sensor_id STRING, ts BIGINT, temperature DOUBLE, dt STRING) USING icebergPARTITIONED BY(dt);-- Append写入1条数据INSERT INTO sensordata VALUES('sensor_01',1635743301,-12.1,'2021-12-01');-- OverWrite写入一条数据INSERT OVERWRITE sensordata VALUES('sensor_02',1635743301,23.6,'2021-12-01');-- Append写入1条数据INSERT INTO sensordata VALUES('sensor_02',1638421701,-22.2,'2021-12-02');-- 删除2021-12-02这一天的数据DELETE FROM sensordata WHERE dt = '2021-12-02';-- Append写入1条数据INSERT INTO sensordata VALUES('sensor_03',1638421701,-22.2,'2021-12-03');
（2）版本信息查询
SELECT s.committed_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY committed_at;committed_at operation snapshot_id is_current_ancestor summary[spark.app.id]2022-01-26 13:26:27.475 append 4815667141509177681 true local-16431741165492022-01-26 13:26:34.404 overwrite 5163041208696993750 true local-16431741165492022-01-26 13:26:40.707 append 2479499780166474991 true local-16431741165492022-01-26 13:26:46.831 delete 6907478957764082176 true local-16431741165492022-01-26 13:26:52.259 append 4719601493924525234 true local-1643174116549
2.1.2 版本回退功能2.1.2.1 rollback_to_snapshot （1）功能描述
（2）查询目前数据情况
spark-sql> SELECT * FROM sensordata;
（3）将sensordata表回退到指定的snapshot_id
我们将表数据回退到delete操作之后的快照Id（6907478957764082176）
CALL hadoop_catalog.system.rollback_to_snapshot('db.sensordata', 6907478957764082176);
（4）验证数据
spark-sql> SELECT * FROM sensordata;2.1.2.2 rollback_to_timestamp 将Iceberg表数据回退到指定的时间戳。
（1）将sensordata表回退到指定的timestamp
我们将表数据回退到delete之前的时间戳（2022-01-26 13:26:46.831）
CALL hadoop_catalog.system.rollback_to_timestamp('db.sensordata', TIMESTAMP '2022-01-26 13:26:46.831') （2）验证数据
SELECT * FROM sensordata;
2.1.3 设置当前版本 与版本回退功能不同的是，这个操作不要求快照是当前表状态的祖先。
（1）将sensordata表设置到指定的snapshot_id
我们将表数据最后一次append操作之后的快照Id（4719601493924525234）
CALL hadoop_catalog.system.set_current_snapshot('db.sensordata', 4719601493924525234)
（2）验证数据
SELECT * FROM sensordata;
2.2 元数据管理 可以使用Procedures管理元数据。
2.2.1 expire_snapshots Iceberg表的每次write/update/delete/upsert/compaction操作，都会生成一个新的Snapshot。用于记录元数据信息和做Time Travel查询。expire_snapshots可用来做删除不必要的快照信息。
（1）删除所有比2022-01-26 13:26:40.707早的所有快照，并且只保留3个快照
CALL hadoop_catalog.system.expire_snapshots('db.sensordata', TIMESTAMP '2022-01-26 13:26:40.707',3);
（2）验证数据，目前已经不存在比2022-01-26 13:26:40.707早的快照了
SELECT s.committed_at, s.operation, h.snapshot_id, h.is_current_ancestor, s.summary['spark.app.id'] FROM hadoop_catalog.db.sensordata.history h JOIN hadoop_catalog.db.sensordata.snapshots s ON h.snapshot_id = s.snapshot_idORDER BY committed_at;2.2.2 rewrite_manifests 重新Manifest文件可以优化查询计划。
（1）重新Manifest文件
CALL hadoop_catalog.system.rewrite_manifests('db.sensordata'); （2）重新Manifest文件，避免占用执行中的Executor内存
CALL hadoop_catalog.system.rewrite_manifests('db.sensordata', false);三 迁移Hive表到Iceberg中3.1 Hive表准备工作 （1）首先创建几张Hive内部表用于测试。
-- 创建测试表1create table ods_sensordata1( sensor_id STRING, ts BIGINT, temperature double)stored as PARQUET location '/dw/hive/ods_sensordata1';-- 创建测试表2create table ods_sensordata2( sensor_id STRING, ts BIGINT, temperature double)stored as PARQUET location '/dw/hive/ods_sensordata2';-- 写数据insert into ods_sensordata1 select 'sensor_20', 1631349207, -23.2;insert into ods_sensordata1 select 'sensor_21', 1631349207, -33.3;insert into ods_sensordata2 select 'sensor_01', 1631349389, -23.4;insert into ods_sensordata2 select 'sensor_02', 1631349390, -33.5; （2）创建Hive外部表
-- 创建测试表1create external table ods_ext_sensordata1( sensor_id STRING, ts BIGINT, temperature double)stored as PARQUET location '/dw/hive/external/ods_ext_sensordata1';-- 创建测试表2create external table ods_ext_sensordata2( sensor_id STRING, ts BIGINT, temperature double)stored as PARQUET location '/dw/hive/external/ods_ext_sensordata2';-- 写数据insert into ods_ext_sensordata1 select 'sensor_20', 1631349207, -23.2;insert into ods_ext_sensordata1 select 'sensor_21', 1631349207, -33.3;insert into ods_ext_sensordata2 select 'sensor_01', 1631349389, -23.4;insert into ods_ext_sensordata2 select 'sensor_02', 1631349390, -33.5;3.2 迁移到Iceberg表中 （1）可以把Hive表替换成Iceberg表，表结构、分区、属性和位置将从源表进行复制，目前支持的表格式有Avro、Parquet和ORC格式。
（2）设置spark-defaults.conf
在spark-defaults.conf配置以下内容，主要用于做表的快照，以及迁移Hive表到Iceberg表
vi conf/spark-defaults.conf spark.sql.catalog.spark_catalog = org.apache.iceberg.spark.SparkSessionCatalogspark.sql.catalog.spark_catalog.type = hive3.2.1 迁移Hive内部表 （1）用下面的命令迁移Hive内部表
CALL hadoop_catalog.system.migrate('spark_catalog.hive_db.ods_sensordata1');
（2）验证迁移效果
spark-sql> select * from spark_catalog.hive_db.ods_sensordata1;
3.2.2 迁移Hive的外部表 （1）用下面的命令迁移Hive外部表
CALL hadoop_catalog.system.migrate('spark_catalog.hive_db.ods_ext_sensordata1');
（2）验证迁移效果
spark-sql> SELECT * FROM spark_catalog.hive_db.ods_ext_sensordata1;四 总结
花粉社群VIP加油站
关于作者
猜你喜欢
