Iceberg

作者：吴文池 背景 iceberg提供合并小文件性能，能够依照用户的配置文件，把多个合乎配置的小文件合并成多个大文件。该流程次要是对源数据做了一次复制。 但在理论生产环境中，数据是始终在变动的，有可能会呈现这种状况：在还未实现数据合并工作时，对之前的数据做出了批改，这就导致正在合并的数据与新数据可能会产生抵触。 那么iceberg是如何解决这种数据抵触的呢？ iceberg行级删除 iceberg的行级批改次要是通过行级删除记录再加上数据记录实现的，所以上面先说一下iceberg行级删除的实现。 iceberg行级删除实现概要 iceberg的行级更新和删除，次要是通过减少delete记录做到的。iceberg总共有两种delete类型，一种是position delete，一种是equality delete，次要区别是在于该数据的插入和批改操作是否在同一个checkpoint内。此次剖析的抵触场景次要是不同快照间的，所以以下阐明都以equality delete为背景，简略说下流程（非upsert模式）。 先在 writer 中设置 equalityFieldColumns : DataStreamSink<Void> append = FlinkSink.forRowData(input) .tableLoader(tableLoader).overwrite(false).distributionMode(DistributionMode.HASH).equalityFieldColumns(Lists.newArrayList("column1", "column2")) .append();设置好之后，writer就会依据设置的列对数据做写入，如果是插入的数据，则写为数据记录，如果是删除数据，则写为删除记录（非upsert模式）： case INSERT:case UPDATE_AFTER: if (upsert) { writer.delete(row);} writer.write(row); break;case UPDATE_BEFORE: if (upsert) { break; // UPDATE_BEFORE is not necessary for UPDATE, we do nothing to prevent delete one row twice} writer.delete(row); break;case DELETE: writer.delete(row); break;在读取数据的时候，会依据数据生成的程序（即sequenceNumber），判断该数据记录是否有对应的删除记录，如果有，则认为该数据曾经被删除，如果没有，则认为该数据仍然存在。 利用删除记录的判断条件 iceberg中应用接口FileScanTask来记录以后扫描工作： public interface FileScanTask extends ScanTask { /** The {@link DataFile file} to scan. *@return the file to scan */DataFile file(); ...

4月16日，由DEEPNOVA开发者社区与Iceberg社区独特推出的“DEEPNOVA MEETUP Online”流动顺利举办。 以《基于 Iceberg 打造实时数据湖》为主题，本次流动邀请了来自阿里云、网易数帆、滴普科技等企业技术专家，独特探讨Iceberg技术发展史及在国产化数据中的利用与实际，瞻望湖仓一体技术将来发展趋势与行业前景。 核心内容： Apache Iceberg过来、现状及将来网易湖仓管理系统ArcticFastData DLink基于Iceberg 构建实时数据湖的优化与实际FastData DLink 对Iceberg索引的优化与实际圆桌探讨：流批一体、湖仓一体的架构劣势与业务价值听众收益： 理解Apache Iceberg的架构原理、个性及利用场景基于Iceberg实现不同性能的技术优化能力及其业务价值凋谢的技术交换社区，与资深技术专家独特探讨湖仓技术精彩照片： DEEPNOVA x Iceberg Meetup Online《基于Iceberg打造实时数据湖》回顾： 【PDF下载】 PC端下载：https://liudao.obs.cn-southwe... 挪动端下载：扫描下方二维码或搜寻 “DEEPEXI滴普科技” 关注微信公众号，回复关键词“PPT”，获取流动材料 【直播回放】https://wx.vzan.com/live/page... DEEPNOVA作为业余的开发者社区，始终致力于促成技术交换、开辟技术视线、建设技术生态，此次流动的顺利开展，也体现出DEEPNOVA踊跃拥抱开源社区的精力，继续做到将前沿技术思维、优质技术内容反哺社区，充沛激发社区创新力。

扫描上图二维码，可间接报名参加线上流动。转发分享直播间邀约参与者，邀约榜前三名还可取得DEEPNOVA专属周边礼盒。快来报名吧！ 4月16日，DEEPNOVA 开发者社区联结Iceberg社区独特推出的“DEEPNOVA MEETUP Online”将正式举办，流动主题为《基于 Iceberg 打造实时数据湖》。 DEEPNOVA 开发者社区是面向技术开发者的交流学习、生态共创平台。滴普科技始终踊跃拥抱开源生态，为Iceberg社区做出突出贡献。 作为此次流动召集人，滴普科技 FastData 产品线总裁杨磊，将与网易数据湖和实时计算团队负责人马进、Apache Iceberg和HBase PMC成员胡争、网易数帆大数据专家范欣欣、滴滴大数据技术总监梁李印、滴普科技DLink产品总经理冯森、滴普科技存储引擎部总监张敢、数据库内核开发工程师简勇华等技术专家，就如何打造实时数据湖等热门话题，为技术爱好者们带来一场技术盛宴。 工夫：2022 年 4 月16 日 14:00 -17:30主题：基于 Iceberg 打造实时数据湖 议程

可能很多人用Presto只用 Hive Connector，其实Iceberg connector跟Hive差不多，不论从实现，还是从性能上都有相互的参照，尤其是在实现方面应用了十分多的Hive connector底层的代码。它创立table也是一样，咱们能够从一个 TPC-DS数据的 customer表里抽几列再创立一个table，你能够指定这个数据的格局，能够是Parquet也能够是ORC格局。也能够同时指定分区（partition），这里用出世的月份这样会容易些，因为月份只有12个，也就是12个分区。咱们创立了这个表之后，跟Hive表一样，你能够select这个表，select* from Iceberg.test，test1是表名，你能够用一个美元符号$加上partitions，这样你能够把这个表的所有分区给列出来。每个分区都会有一个统计，比如说上面第一行8月，能看到它有多少行有多少个文件，大小总共有多大，同时对于customer_sk这一列，能看到最小值多少最大值多少。前面的birth date就是日期，对于8月最小值是1，最大值是31，空值有若干。因为8月是大月，前面的9月是小月最大值是30，每一个partition都会有本人的统计，前面咱们会再讲， predicate pushdown会用到这个，能够让咱们跳过很多的分区，其实Hive也有这个性能，只不过可能有些数据在Hive metastore上，元数据这里没有的话，用不上这个性能，但在Iceberg上它内嵌在table里了，就会比拟好用一些。后面提到Iceberg会有一些事务（transaction）反对。咱们试着往这个表里退出一行， SK是1000，日期是40，我特意插入了一个不可能存在的日期月份是13，这样等于说我新创建了一个partition。其实不论是不是新建partition都会产生一个新的快照（snapshot），在Presto里，通过select * from 表, 表的名字下面加一个美元$符号，而后再加一个snapshots，就能够列出这个 table所有的snapshots。大家能够看到有两个snapshots，因为新建table时出一个，插入一行之后又出一个，manifest list就有两个avro文件，第二个snapshot基于第一个，第二个snapshot 的parent ID就是第一个snapshot的parent ID，待会咱们会用snapshot ID来做time travel。 对于这样一个文件，咱们加了一个partition进去之后会怎么样，看一下这个目录，其实Iceberg的目录非常简单，咱们指定一个目录，它在这上面就创立一个test1，里边有两个文件夹，一个是data (数据)，一个是metadata（元数据）。数据里边是依照月份来分区的，这外面是 14个分区，因为12个月份，还有个空值，再加上咱们新加的月份13，等于当初一共14个分区，这个文件就是这么组织的，而每个目录上面就是parquet文件。那么咱们在query的时候会产生什么呢？ 这个其实大家都会——写个SQL，从这个select*from Test1的时候，指定一个条件，我这个月份是13，那就把我方才新插入的那一条记录给调进去了。我前面会介绍一下怎么做工夫穿梭(Time travel)，咱们能够看到在这个表名test1这有个@符，前面我能够加一个 snapshot ID，如果我用第二个（快照）snapshot，就能查到这个记录，如果我用第一个snapshot就没有这个记录，为什么？因为第一个query产生在插入这条记录之前。这还挺有用的，因为有的时候就想查一下我这个表昨天是什么样的。但这也有问题，如果你频繁插入数据的话，就会产生大量的snapshot，avro外面就会有大量的数据。那咱们是不是要丢掉一些过期的快照？这也是个优化点，当初presto还没有，但当前我感觉咱们会把它做进去。 另外有些敌人会问，既然Iceberg connector有这个性能了，能不能用它来取代MySQL，做OLTP来解决一些在线的transaction数据? ——能够，然而不能像MySQL那么用，频繁的插入数据还会带来一些问题，须要做更深刻的优化，间接这么用的话会产生大量的小文件和快照，但这些都有方法解决，咱们前面会把它缓缓迭代进去。这个是我的前共事Chunxu做 Schema Evolution的时候截的一张图。能够看到这也是Iceberg的一个亮点，就是说这个表原来有几列，我能够加一列或者改一列，当然这也不难，因为原来 Hive table也能够这样做，然而做完之后，你的table还能不能查？Iceberg给咱们的答案是 table改完了还能查，当然这里边也有tricky的中央，外面的数据也不是这么残缺，然而不论怎么样它没出错，你先改好table，用老的query还可能查到。这个性能我感觉还是挺实用的，因为各个公司 table总在改，改完之后 table在presto这边还是能够查的。最近这两三个月，有几个性能进来之后给咱们的一些货色解锁了。 1.第一份credit要重点给亚马逊公司AWS 的Jack Ye，他做了一个 native folder的反对，这个在Iceberg叫做Hadoop catalog，盘活了咱们的很多性能，解决了咱们十分多的痛点。 2.另外就是腾讯的Baolong，他把local cache这个性能给加上去了，当初Iceberg connector能够和 RaptorX那一套，就是Hive connector里的cache，同样享受local cache，失去提速。当然这个不是那么简略，那么开箱即用，可能会须要一些配置，前面我会再具体地讲。 3.接下来就是Uber的Xinli Shang，咱们俩给parquet做了降级。Xinli Shang是Parquet社区的chair，他给parquet做了降级后，咱们拿过去放在presto里，咱们的降级工作历时大略半年，降级到新的parquet之后，咱们也解锁了Iceberg 1.12，有更多新的性能，包含对v2 的Iceberg table的反对。 4.还有一个predicate pushdown，在前面Beinan（Alluxio）也会具体地讲一下，这是能够优化查问的一个性能。 这就是我方才提到的 Jack做的 native catalog——原来在Iceberg叫Hadoop catalog，其实Iceberg数据也是存在 S3、HDFS、GCS里的。它的每一个table上面既有元数据，又有数据，那为什么还须要Hive metastore，还要去Hive metastore里取元数据呢？这是因为最开始的Hive catalog还是要依赖Hive的元数据的，咱们须要找到 table的门路，到这个table里把Iceberg本人的元数据加载进去，而后用 presto进行查问。有了 Jack这个很好的批改，咱们能够反对Hadoop catalog，你间接给它一个门路，table都放在这个门路上面，它到这个门路下来扫一下，就能够录入所有的table，像table1，table2，table3， 每个table多少元数据，咱们就不再须要Hive metastore了。有了这个native catalog之后， presto和Iceberg的联合就残缺了。原本咱们还依赖于一个额定的元数据存储，当初咱们能够间接应用native catalog，这解决了十分多的痛点。 ...

Alluxio是2014年在伯克利 AMPLab孵化的一个我的项目，那时候名叫Tachyon，是跟Spark同一期孵化的分布式存储我的项目。截止到明天为止，咱们这个社区里曾经有超过1000名的contributor参加搭建了社区代码和各种流动，在Slack committee外面曾经有5000以上的 member进行互动，大家也把技术广泛应用在各种开源场景外面。在去年的时候，咱们也被谷歌评比为最具影响力的十大 Java-based的开源社区我的项目，如果大家对这个我的项目感兴趣的话，欢送扫描二维码退出咱们 slack社区，跟咱们进行探讨。当初Alluxio我的项目曾经宽泛地利用在各种不同的大厂和小厂外面，包含互联网企业，云计算的提供商，也包含银行、不同的通信行业的企业，如果大家感兴趣的话，咱们也会进行更多的分享，邀请大家来一起做技术分享，搭建基于Alluxio的技术平台。Alluxio是一个数据编排层，它不仅服务于数据analytics，也服务于AI，咱们最次要的场景也是基于cloud的场景。Alluxio第一个性能是能够应用Alluxio去连贯所有的计算引擎和存储系统，如果你下面的零碎搭载了 Presto、Spark、 MapReduce、Hive这一类的零碎，你能够通过Alluxio去拜访上面所有的存储引擎，包含但不仅限于我在这里列的HDFS，S3、GCS、Azure和NFS等，咱们有更全面的反对，包含下层的计算引擎和上层的存储系统。咱们的指标是把数据从远端的存储带回到近端的计算外面，因为特地是在云计算、混合计算或者是多数据中心的场景下，存储很多时候都不是跟传统的计算在一起的，所以这时候拜访数据就会有很多的问题。在这种场景下，咱们须要把数据从新从远端的存储带回到近端的计算下面达成更高效的拜访。最次要的利用用例有三个： 第一个用例是云计算的场景，云计算场景个别都是用云计算存储，所以它的性能和稳定性并不是十分的好, 把Alluxio加上去当前，能够达到更好的 SLA一致性，更好的性能，并且能够省去流量，比如说私有云存储的流量应用须要花钱，但咱们能够通过caching solution去缓解流量的开销。 第二个场景是混合云计算，比如说咱们有一部分的数据是放在私有云外面，有一部分数据还放在传统的公有云外面，如果公有云的数据想去拜访私有云，它的带宽是十分受限的，这个时候能够把Alluxio加进去，成为一个缓存层，来缓解这个问题，这就是混合云部署。 第三个场景是一个多计算中心解决方案，比方你在北京、上海，兰州都有一个数据中心，能够在这种场景下部署Alluxio去减速利用场景。咱们的倒退次要有上面三个方向，咱们始终强调Alluxio尽管是一个缓存层的组件，但它不仅仅是一个缓存层。 第一个组件就是unify data lake，能够了解为反对不同的API向下层的计算引擎，也能够向下反对不同的存储系统，你只有治理好 Alluxio的一套 API，就能够治理好整套的data lake solution。 第二个是能够更无效地获取数据并治理数据，这也是缓存层次要的性能，次要是给业务进行减速。咱们也提供data policy引擎，比如说你想把数据从一个HDFS集群挪动到另一个HDFS集群，能够通过policy engine定时并且定量地进行数据挪动，从而更好地治理数据集群。 第三个是多云部署。拿美国这边打个比方，比如说你须要用AWS的 S3, 同时要用GCP的BigQuery引擎，你能够进行混合云的部署，也就是说你能够在 GCP外面部署Alluxio集群去query S3的数据，这样就能够更好地搭建混合云。 举例在国内比拟常见的场景是从公有云的HDFS到混合云的对象存储，其中咱们提供几个性能，第一个叫unified name space，你能够把 HDFS集群和对象存储的集群都mount到一个对立的Alluxio cluster外面进行治理，提供残缺的基于对象存储的一套剖析解决方案。在这种场景下，比拟常见到的是混合云部署，就是说既有私有云也有公有云。咱们来看一下部署的架构，在这个场景里，有两个不同的私有云提供商，一个是AWS，一个是Google Cloud。在AWS和Google Cloud外面咱们都部署了Alluxio集群。其中第三个集群是一个private cluster，就是公有云的集群，咱们也同时部署了Alluxio，它的底层其实是一个MINIO集群，也就是说MINIO的集群同时提供了数据服务给两个私有云和一个公有云，而在这种场景下，你能够用Alluxio作为缓存减速提供给不同的下层计算引擎，包含在AWS上的 EMR Google cloud dataproc和包含在公有云上面本人搭建的Hive， Spark和Presto场景，你能够达成一个残缺的循环，不须要用公有云外面的MINIO提供非常低的性能，去给不同的私有云和公有云计算引擎。 混合云到多个云的场景大家能够看到，同样是两个私有云提供商，右边是Microsoft Azure，左边是AWS，能够看到上面ETL job和Ingestion都会间接写到Alluxio外面，在这种场景下，你能够抉择写入Alluxio的形式，把这个数据写回到底层的 HDFS下面去， 而后Alluxio会把这个数据分享给不同的数据中心，包含左边的HDFS集群和下面的几个私有云的Alluxio集群。在这种搭建下，只须要一个Alluxio集群，就能够实现数据共享，不须要搭建 ETL数据通道，再去把所有的数据写回到不同的HDFS 集群和另外的Alluxio或者是云计算的对象存储外面。Iceberg是一种开放式的table format，它次要是作为对标 Hive的一种文件，是一种表格存储的格局，次要的应答场景是比拟大型的数据分析场景。它提供几种不同的性能，次要有 Schema evolution，hidden partitioning，partition layout evolution 和time travel，还提供version rollback的性能，很多性能是专门为了满足美国对于数据隐衷法案的需要，提供事务性的反对。绝对于Hive很难实现事务性的反对，Apache Iceberg提供了更好的反对。Iceberg的schema evolution都是基于元数据操作的，没有任何的数据操作，能够减少column，drop column，能够rename column，也能够更新 column struct field，map key，map value或者 list element，甚至能够reorder，change the order of column or field in a nested struct。其实他们当初也反对z-ordering这种格局，为了让数据能失去更好的减速成果。那么它是怎么实现事务性反对呢？它用了比拟传统的数据库的理念，就是snapshot，它的不同的表格版本，是用snapshot来爱护的，snapshot上面有一系列的manifest file来表征这个表在某一个非凡的工夫点，比如说在11月27号，它是一个 snapshot，有可能过比如说一个小时或者是一天，它就更新一个snapshot，所以每个表格是通过snapshot去保障它的不同的版本的数据的一致性。接下来咱们来看一下为什么 Iceberg提出了这种计划。 ...