调研iceberg的使用,记录下flink 集成iceberg的坑(持续更新)
简单来说 iceberg可以看作hdfs的一个文件格式,hive的升级版,提供sql操作的kafka升级版,可以用来做实时数仓,分钟级别
目前测试1.3x flink with 0.13.x iceberg,stream api 可以正常读kafka数据源 写/读iceberg(hive/hadoop catalog)
目前测试1.11x flink with 0.11.x iceberg,stream/table api 均可正常读kafka数据源 写/读iceberg(hive/hadoop catalog)
目前测试1.3x 和1.11 数据可以互通,即使用1.13版本的stream 插入数据,如需使用sql则可以使用1.1x版本
前提:在小文件合并程序中,对data文件进行合并,并且只保留一个快照,flink 读取iceberg报错
原因:iceberg对数据的每一次写,都会产生快照,在程序中对data文件进行合并后,快照时间戳为最新,但是其他data还被其他的快照所引用,元数据找不到快照报错
解决:删除一段时间前即可
Snapshot snapshot = table.currentSnapshot();
// 删除一个小时前的
long l = snapshot.timestampMillis() - 3600000;
table.expireSnapshots().expireOlderThan(l).commit();
因篇幅问题不能全部显示,请点此查看更多更全内容