1. expire过期处理的作用

expire过期的snapshot-id，就不会出现在新的vN.metadata.json文件中

2. 查看表snapshot-log

查看最新版本的vN.metadata.json，显示的snapshot-log部分如下：

"snapshot-log" : [ {"timestamp-ms" : 1645089884916,"snapshot-id" : 3105878303282846379}, {"timestamp-ms" : 1645089898911,"snapshot-id" : 781839495286765092}, {"timestamp-ms" : 1645089917661,"snapshot-id" : 242516093407225541}, {"timestamp-ms" : 1645089922843,"snapshot-id" : 6992061996353419515}, {"timestamp-ms" : 1645090105742,"snapshot-id" : 4375332003125072899}, {"timestamp-ms" : 1645090219342,"snapshot-id" : 4792595715782867813} ]

3. expire过期程序

3.1 pom.xml

添加的依赖部分如下：

        <dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.3.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs-client</artifactId><version>3.3.1</version></dependency><dependency><groupId>org.apache.avro</groupId><artifactId>avro</artifactId><version>1.10.1</version></dependency><dependency><groupId>org.apache.iceberg</groupId><artifactId>iceberg-core</artifactId><version>0.13.0</version></dependency>

3.2 放置hdfs-site.xml

将hdfs-site.xml放到项目的resources目录下

3.3 flink_test.scala

import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.Table
import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.hadoop.HadoopCatalogobject flink_test {def main(args: Array[String]): Unit = {// =======================expire过期快照========================val icebergWarehousePath = "hdfs://nnha/user/iceberg/warehouse"val expireOlderThanTimestamp = 1645089917661Lval conf = new Configuration()// 添加DistributedFileSystem类，否则会报异常：org.apache.hadoop.fs.UnsupportedFileSystemException: No FileSystem for scheme "hdfs"conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem")val hadoopCatalog = new HadoopCatalog(conf, icebergWarehousePath)val table: Table = hadoopCatalog.loadTable(TableIdentifier.of("iceberg_db", "my_user"))table.expireSnapshots().expireOlderThan(expireOlderThanTimestamp).commit()}
}

4. 运行程序

将程序进行打包，上传到服务器，然后运行

[root@flink1 ~]# java -classpath flink_dev-1.0-SNAPSHOT.jar flink_test
log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[root@flink1 ~]#

5. 查看结果

再次查看最新版本的vN.metadata.json，显示的snapshot-log部分如下：

  "snapshot-log" : [ {"timestamp-ms" : 1645089917661,"snapshot-id" : 242516093407225541}, {"timestamp-ms" : 1645089922843,"snapshot-id" : 6992061996353419515}, {"timestamp-ms" : 1645090105742,"snapshot-id" : 4375332003125072899}, {"timestamp-ms" : 1645090219342,"snapshot-id" : 4792595715782867813} ]

可以看到，timestamp-ms小于1645089917661的snapshot-id，不会出现在新的metadata文件中了

使用Java/Scala API对Iceberg数据湖进行snapshot快照expire过期处理相关推荐

Flink 如何实时分析 Iceberg 数据湖的 CDC 数据
简介:数据湖的架构中,CDC 数据实时读写的方案和原理本文由李劲松.胡争分享,社区志愿者杨伟海.李培殿整理.主要介绍在数据湖的架构中,CDC 数据实时读写的方案和原理.文章主要分为 4 个部分内容: ...
Apache Iceberg 数据湖从入门到放弃(2) —— 初步入门
在介绍如何使用Iceberg之前,先简单地介绍一下Iceberg catalog的概念.catalog是Iceberg对表进行管理(create.drop.rename等)的一个组件.目前Iceber ...
winfrom里面使用类似于table的合并的控件_Flink集成iceberg数据湖之合并小文件
背景使用流式数据入湖开启压缩程序快照过期删除无用文件数据查询遇到的坑最大并发度问题文件被重复压缩扫描任务读取文件问题不读取大文件优化生成CombinedScanTask 后续问 ...
java sqoop api 导mysql数据到hdfs
package com.example.demo.controller;import org.apache.hadoop.conf.Configuration; import org.apache.s ...
Apache Iceberg数据湖与Trino的集成
目录 1. 介绍 2. 配置Connector 1. 介绍 Trino连接Iceberg的Connector,需要通过Hive的metastore,查询每个partition的HDFS locatio ...
Flink重写Iceberg数据湖小文件变大文件
1. 重写小文件变大文件 Flink支持Batch任务,将iceberg表的小文件重写成大文件合并前HDFS的metadata和data目录文件如下: [root@flink1 ~]# [root@ ...
Flink + Iceberg + 对象存储，构建数据湖方案
简介:上海站 Flink Meetup 分享内容,如何基于Flink.对象存储.Iceberg 来构建数据湖生态. 本文整理自 Dell 科技集团高级软件研发经理孙伟在 4 月 17 日上海站 Fl ...
数据湖Iceberg | 实时数据仓库的发展、架构和趋势
数据处理现状:当前基于Hive的离线数据仓库已经非常成熟,数据中台体系也基本上是围绕离线数仓进行建设.但是随着实时计算引擎的不断发展以及业务对于实时报表的产出需求不断膨胀,业界最近几年就一直聚焦并探索 ...
大数据架构师——数据湖技术（二）
文章目录数据湖技术数据湖技术之Iceberg Spark 与 Iceberg 整合 1. Spark3.2.1 与 Iceberg0.13.2整合添加依赖 Spark 设置 Catalog 配置 ...

使用Java/Scala API对Iceberg数据湖进行snapshot快照expire过期处理

目录

1. expire过期处理的作用

2. 查看表snapshot-log

3. expire过期程序

3.1 pom.xml

3.2 放置hdfs-site.xml

3.3 flink_test.scala

4. 运行程序

5. 查看结果

使用Java/Scala API对Iceberg数据湖进行snapshot快照expire过期处理相关推荐

最新文章

热门文章