基于Flink1.14 + Iceberg0.13构建实时数据湖实战

王知无

大厂在职,学习规划/简历精修请私信。

点击上方 蓝色字体 ，选择“设为星标”

回复" 面试" 获取更多惊喜

八股文教给我，你们专心刷题和面试

Hi，我是王知无，一个大数据领域的原创作者。
放心关注我，获取更多行业的一手消息。

Flink SQL Client配置Iceberg
Java/Scala pom.xml配置
Catalog
3.1 Hive Catalog
3.2 HDFS Catalog
数据库和表相关DDL命令
4.1 创建数据库
4.2 创建表(不支持primary key等)
4.3 修改表
4.4 删除表
插入数据到表
5.1 insert into
5.2 insert overwrite(只有Batch模式支持，且overwrite粒度为partition)
查询数据
暂时还不支持通过Flink SQL读取Iceberg表的元数据，可以通过Java API读取

1. Flink SQL Client配置Iceberg

Flink集群需要使用Scala 2.12版本的

将Iceberg的依赖包下载放到Flink集群所有服务器的lib目录下，然后重启Flink


             [root@flink1~]#wget-P/root/flink-1.14.3/libhttps://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.14/0.13.0/iceberg-flink-runtime-1.14-0.13.0.jar
[root@flink1~]#
[root@flink1~]#scp/root/flink-1.14.3/lib/iceberg-flink-runtime-1.14-0.13.0.jarroot@flink2:/root/flink-1.14.3/lib
iceberg-flink-runtime-1.14-0.13.0.jar100%23MB42.9MB/s00:00
[root@flink1~]#scp/root/flink-1.14.3/lib/iceberg-flink-runtime-1.14-0.13.0.jarroot@flink3:/root/flink-1.14.3/lib
iceberg-flink-runtime-1.14-0.13.0.jar100%23MB35.4MB/s00:00
[root@flink1~]#

Iceberg默认支持Hadoop Catalog。如果需要使用Hive Catalog，需要将flink-sql-connector-hive-3.1.2_2.12-1.14.3.jar放到Flink集群所有服务器的lib目录下，然后重启Flink

然后启动SQL Client就可以了

2. Java/Scala pom.xml配置

添加如下依赖


             <dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-flink</artifactId>
<version>0.13.0</version>
<scope>provided</scope>
</dependency>

3. Catalog

3.1 Hive Catalog

注意：测试的时候，从Hive中查询表数据，查询不到。但是从Trino查询可以查询到数据

使用Hive的metastore保存元数据，HDFS保存数据库表的数据


             FlinkSQL>createcataloghive_catalogwith(
>'type'='iceberg',
>'catalog-type'='hive',
>'property-version'='1',
>'cache-enabled'='true',
>'uri'='thrift://hive1:9083',
>'client'='5',
>'warehouse'='hdfs://nnha/user/hive/warehouse',
>'hive-conf-dir'='/root/flink-1.14.3/hive_conf'
[INFO]Executestatementsucceed.
FlinkSQL>

property-version: 为了向后兼容，以防property格式改变。当前设置为1即可
cache-enabled: 是否开启catalog缓存，默认开启
clients: 在hive metastore中，hive_catalog供客户端访问的连接池大小，默认是2
warehouse: 是Flink集群所在的HDFS路径, hive_catalog下的数据库表存放数据的位置
hive-conf-dir: hive集群的配置目录。只能是Flink集群的本地路径，从hive-site.xml解析出来的HDFS路径，是Flink集群所在HDFS路径
warehouse的优先级比hive-conf-dir的优先级高

3.2 HDFS Catalog

用HDFS保存元数据和数据库表的数据。warehouse是Flink集群所在的HDFS路径


             FlinkSQL>createcataloghadoop_catalogwith(
>'type'='iceberg',
>'catalog-type'='hadoop',
>'property-version'='1',
>'cache-enabled'='true',
>'warehouse'='hdfs://nnha/user/iceberg/warehouse'
[INFO]Executestatementsucceed.
FlinkSQL>

通过配置conf/sql-cli-defaults.yaml实现永久catalog。但测试的时候并未生效


             [root@flink1~]#cat/root/flink-1.14.3/conf/sql-cli-defaults.yaml
catalogs:
-name:hadoop_catalog
type:iceberg
catalog-type:hadoop
property-version:1
cache-enabled:true
warehouse:hdfs://nnha/user/iceberg/warehouse
[root@flink1~]#
[root@flink1~]#chown501:games/root/flink-1.14.3/conf/sql-cli-defaults.yaml

下面我们重点以Hadoop Catalog为例，进行测试讲解

4. 数据库和表相关DDL命令

4.1 创建数据库

Catalog下面默认都有一个default数据库


             FlinkSQL>createdatabasehadoop_catalog.iceberg_db;
[INFO]Executestatementsucceed.
FlinkSQL>usehadoop_catalog.iceberg_db;
[INFO]Executestatementsucceed.
FlinkSQL>

会在HDFS目录上创建iceberg_db子目录
如果删除数据库，会删除HDFS上的iceberg_db子目录

4.2 创建表(不支持primary key等)


             FlinkSQL>createtablehadoop_catalog.iceberg_db.my_user(
>user_idbigintcomment'用户ID',
>user_namestring,
>birthdaydate,
>countrystring
>)comment'用户表'
>partitionedby(birthday,country)with(
>'write.format.default'='parquet',
>'write.parquet.compression-codec'='gzip'
[INFO]Executestatementsucceed.
FlinkSQL>

目前表不支持计算列、primay key, Watermark
不支持计算分区。但是iceberg支持计算分区
创建表生成的文件信息如下：


             [root@flink1~]#
[root@flink1~]#hadoopfs-lshdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata
Found2items
-rw-r--r--1rootsupergroup21152022-02-1322:01hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v1.metadata.json
-rw-r--r--1rootsupergroup12022-02-1322:01hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/version-hint.text
[root@flink1~]#

查看v1.metadata.json，可以看到"current-snapshot-id" : -1


             FlinkSQL>createtablehadoop_catalog.iceberg_db.my_user_copy
>likehadoop_catalog.iceberg_db.my_user;
[INFO]Executestatementsucceed.
FlinkSQL>

复制的表拥有相同的表结构、分区、表属性

4.3 修改表

修改表属性


             FlinkSQL>altertablehadoop_catalog.iceberg_db.my_user_copy
>'write.format.default'='avro',
>'write.avro.compression-codec'='gzip'
[INFO]Executestatementsucceed.
FlinkSQL>

目前Flink只支持修改iceberg的表属性

重命名表


             FlinkSQL>altertablehadoop_catalog.iceberg_db.my_user_copy
>renametohadoop_catalog.iceberg_db.my_user_copy_new;
[ERROR]CouldnotexecuteSQLstatement.Reason:
java.lang.UnsupportedOperationException:CannotrenameHadooptables
FlinkSQL>

Hadoop Catalog中的表不支持重命名表

4.4 删除表


             FlinkSQL>droptablehadoop_catalog.iceberg_db.my_user_copy;
[INFO]Executestatementsucceed.
FlinkSQL>

会删除HDFS上的my_user_copy子目录

5. 插入数据到表

5.1 insert into

insert into … values …
insert into … select …


             FlinkSQL>insertintohadoop_catalog.iceberg_db.my_user(
>user_id,user_name,birthday,country
>)values(1,'zhang_san',date'2022-02-01','china'),
>(2,'li_si',date'2022-02-02','japan');
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:f1aa8bee0be5bda8b166cc361e113268
FlinkSQL>
FlinkSQL>insertintohadoop_catalog.iceberg_db.my_userselect(user_id+1),user_name,birthday,countryfromhadoop_catalog.iceberg_db.my_user;
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:c408e324ca3861b39176c6bd15770aca
FlinkSQL>

HDFS目录结果如下


             hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/data/birthday=2022-02-01/country=china/00000-0-4ef3835f-b18b-4c48-b47a-85af1771a10a-00001.parquet
hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/data/birthday=2022-02-01/country=china/00000-0-6e66c02b-cb09-4fd0-b669-15aa7f5194e4-00001.parquet
hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/data/birthday=2022-02-02/country=japan/00000-0-4ef3835f-b18b-4c48-b47a-85af1771a10a-00002.parquet
hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/data/birthday=2022-02-02/country=japan/00000-0-6e66c02b-cb09-4fd0-b669-15aa7f5194e4-00002.parquet

5.2 insert overwrite(只有Batch模式支持，且overwrite粒度为partition)

只支持Flink Batch模式，不支持Streaming模式

insert overwrite替换多个整个分区，而不是一行数据。如果不是分区表，则替换的是整个表，如下所示：


             FlinkSQL>set'execution.runtime-mode'='batch';
[INFO]Sessionpropertyhasbeenset.
FlinkSQL>
FlinkSQL>insertoverwritehadoop_catalog.iceberg_db.my_uservalues(4,'wang_wu',date'2022-02-02','japan');
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:63cf6c27060ec9ebdce75b785cc3fa3a
FlinkSQL>set'sql-client.execution.result-mode'='tableau';
[INFO]Sessionpropertyhasbeenset.
FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user;
+---------+-----------+------------+---------+
|user_id|user_name|birthday|country|
+---------+-----------+------------+---------+
|1|zhang_san|2022-02-01|china|
|4|wang_wu|2022-02-02|japan|
|2|zhang_san|2022-02-01|china|
+---------+-----------+------------+---------+
3rowsinset

birthday=2022-02-02/country=japan分区下的数据如下，insert overwrite也是新增一个文件


             birthday=2022-02-02/country=japan/00000-0-1d0ff907-60a7-4062-93a3-9b443626e383-00001.parquet
birthday=2022-02-02/country=japan/00000-0-4ef3835f-b18b-4c48-b47a-85af1771a10a-00002.parquet
birthday=2022-02-02/country=japan/00000-0-6e66c02b-cb09-4fd0-b669-15aa7f5194e4-00002.parquet

insert ovewrite … partition替换指定分区


             FlinkSQL>insertoverwritehadoop_catalog.iceberg_db.my_userpartition(birthday='2022-02-02',country='japan')select5,'zhao_liu';
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:97e9ba4131028c53461e739b34108ae0
FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user;
+---------+-----------+------------+---------+
|user_id|user_name|birthday|country|
+---------+-----------+------------+---------+
|1|zhang_san|2022-02-01|china|
|5|zhao_liu|2022-02-02|japan|
|2|zhang_san|2022-02-01|china|
+---------+-----------+------------+---------+
3rowsinset
FlinkSQL>

6. 查询数据

Batch模式


             FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user;
+---------+-----------+------------+---------+
|user_id|user_name|birthday|country|
+---------+-----------+------------+---------+
|1|zhang_san|2022-02-01|china|
|5|zhao_liu|2022-02-02|japan|
|2|zhang_san|2022-02-01|china|
+---------+-----------+------------+---------+
3rowsinset
FlinkSQL>

streaming模式

查看最新的snapshot-id


             [root@flink1conf]#hadoopfs-cathdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/version-hint.text

我们前面创建表 + 两次insert + 两次insert overwrite，所以最新的版本号为5。然后我们查看该版本号对于的metadata json文件


             [root@flink1~]#hadoopfs-cathdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v5.metadata.json
"format-version":1,
"table-uuid":"84a5e90d-7ae9-4dfd-aeab-c74f07447513",
"location":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user",
"last-updated-ms":1644761481488,
"last-column-id":4,
"schema":{
"type":"struct",
"schema-id":0,
"fields":[{
"id":1,
"name":"user_id",
"required":false,
"type":"long"
"id":2,
"name":"user_name",
"required":false,
"type":"string"
"id":3,
"name":"birthday",
"required":false,
"type":"date"
"id":4,
"name":"country",
"required":false,
"type":"string"
"current-schema-id":0,
"schemas":[{
"type":"struct",
"schema-id":0,
"fields":[{
"id":1,
"name":"user_id",
"required":false,
"type":"long"
"id":2,
"name":"user_name",
"required":false,
"type":"string"
"id":3,
"name":"birthday",
"required":false,
"type":"date"
"id":4,
"name":"country",
"required":false,
"type":"string"
"partition-spec":[{
"name":"birthday",
"transform":"identity",
"source-id":3,
"field-id":1000
"name":"country",
"transform":"identity",
"source-id":4,
"field-id":1001
"default-spec-id":0,
"partition-specs":[{
"spec-id":0,
"fields":[{
"name":"birthday",
"transform":"identity",
"source-id":3,
"field-id":1000
"name":"country",
"transform":"identity",
"source-id":4,
"field-id":1001
"last-partition-id":1001,
"default-sort-order-id":0,
"sort-orders":[{
"order-id":0,
"fields":[]
"properties":{
"write.format.default":"parquet",
"write.parquet.compression-codec":"gzip"
"current-snapshot-id":138573494821828246,
"snapshots":[{
"snapshot-id":8012517928892530314,
"timestamp-ms":1644761130111,
"summary":{
"operation":"append",
"flink.job-id":"8f228ae49d34aafb4b2887db3149e3f6",
"flink.max-committed-checkpoint-id":"9223372036854775807",
"added-data-files":"2",
"added-records":"2",
"added-files-size":"2487",
"changed-partition-count":"2",
"total-records":"2",
"total-files-size":"2487",
"total-data-files":"2",
"total-delete-files":"0",
"total-position-deletes":"0",
"total-equality-deletes":"0"
"manifest-list":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/snap-8012517928892530314-1-5c33451b-48ab-4ce5-be7a-2c2d2dc9e11d.avro",
"schema-id":0
"snapshot-id":453371561664052237,
"parent-snapshot-id":8012517928892530314,
"timestamp-ms":1644761150082,
"summary":{
"operation":"append",
"flink.job-id":"813b7a17c21ddd003e1a210b1366e0c5",
"flink.max-committed-checkpoint-id":"9223372036854775807",
"added-data-files":"2",
"added-records":"2",
"added-files-size":"2487",
"changed-partition-count":"2",
"total-records":"4",
"total-files-size":"4974",
"total-data-files":"4",
"total-delete-files":"0",
"total-position-deletes":"0",
"total-equality-deletes":"0"
"manifest-list":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/snap-453371561664052237-1-bc0e56ec-9f78-4956-8412-4d8ca70ccc19.avro",
"schema-id":0
"snapshot-id":6410282459040239217,
"parent-snapshot-id":453371561664052237,
"timestamp-ms":1644761403566,
"summary":{
"operation":"overwrite",
"replace-partitions":"true",
"flink.job-id":"f7085f68e5ff73c1c8aa1f4f59996068",
"flink.max-committed-checkpoint-id":"9223372036854775807",
"added-data-files":"1",
"deleted-data-files":"2",
"added-records":"1",
"deleted-records":"2",
"added-files-size":"1244",
"removed-files-size":"2459",
"changed-partition-count":"1",
"total-records":"3",
"total-files-size":"3759",
"total-data-files":"3",
"total-delete-files":"0",
"total-position-deletes":"0",
"total-equality-deletes":"0"
"manifest-list":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/snap-6410282459040239217-1-2b20c57e-5428-4483-9f7b-928b980dd50d.avro",
"schema-id":0
"snapshot-id":138573494821828246,
"parent-snapshot-id":6410282459040239217,
"timestamp-ms":1644761481488,
"summary":{
"operation":"overwrite",
"replace-partitions":"true",
"flink.job-id":"d434d6d4f658d61732d7e9a0a85279fc",
"flink.max-committed-checkpoint-id":"9223372036854775807",
"added-data-files":"1",
"deleted-data-files":"1",
"added-records":"1",
"deleted-records":"1",
"added-files-size":"1251",
"removed-files-size":"1244",
"changed-partition-count":"1",
"total-records":"3",
"total-files-size":"3766",
"total-data-files":"3",
"total-delete-files":"0",
"total-position-deletes":"0",
"total-equality-deletes":"0"
"manifest-list":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/snap-138573494821828246-1-b243b39e-7122-4571-b6fa-c902241e36a8.avro",
"schema-id":0
"snapshot-log":[{
"timestamp-ms":1644761130111,
"snapshot-id":8012517928892530314
"timestamp-ms":1644761150082,
"snapshot-id":453371561664052237
"timestamp-ms":1644761403566,
"snapshot-id":6410282459040239217
"timestamp-ms":1644761481488,
"snapshot-id":138573494821828246
"metadata-log":[{
"timestamp-ms":1644760911017,
"metadata-file":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v1.metadata.json"
"timestamp-ms":1644761130111,
"metadata-file":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v2.metadata.json"
"timestamp-ms":1644761150082,
"metadata-file":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v3.metadata.json"
"timestamp-ms":1644761403566,
"metadata-file":"hdfs://nnha/user/iceberg/warehouse/iceberg_db/my_user/metadata/v4.metadata.json"
}[root@flink1~]#

可以看到 "current-snapshot-id" : 138573494821828246,，表示当前的snapshot-id


             FlinkSQL>set'execution.runtime-mode'='streaming';
[INFO]Sessionpropertyhasbeenset.
FlinkSQL>
FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user
>/*+options(
>'streaming'='true',
>'monitor-interval'='5s'
+----+----------------------+--------------------------------+------------+--------------------------------+
|op|user_id|user_name|birthday|country|
+----+----------------------+--------------------------------+------------+--------------------------------+
|+I|5|zhao_liu|2022-02-02|japan|
|+I|2|zhang_san|2022-02-01|china|
|+I|1|zhang_san|2022-02-01|china|

可以看到最新snapshot对应的数据


             FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user
>/*+options(
>'streaming'='true',
>'monitor-interval'='5s',
>'start-snapshot-id'='138573494821828246'
+----+----------------------+--------------------------------+------------+--------------------------------+
|op|user_id|user_name|birthday|country|
+----+----------------------+--------------------------------+------------+--------------------------------+

这里只能指定最后一个insert overwrite操作的snapshot id，及其后面的snapshot id，否则后台会报异常，且程序一直处于restarting的状态：


             java.lang.UnsupportedOperationException:Foundoverwriteoperation,cannotsupportincrementaldatainsnapshots(8012517928892530314,138573494821828246]

在本示例中snapshot id: 138573494821828246，是最后一个snapshot id，同时也是最后一个insert overwrite操作的snapshot id。如果再insert两条数据，则只能看到增量的数据


             FlinkSQL>insertintohadoop_catalog.iceberg_db.my_user(
>user_id,user_name,birthday,country
>)values(6,'zhang_san',date'2022-02-01','china');
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:8eb279e61aed66304d78ad027eaf8d30
FlinkSQL>insertintohadoop_catalog.iceberg_db.my_user(
>user_id,user_name,birthday,country
>)values(7,'zhang_san',date'2022-02-01','china');
[INFO]SubmittingSQLupdatestatementtothecluster...
[INFO]SQLupdatestatementhasbeensuccessfullysubmittedtothecluster:
JobID:70a050e455d188d0d3f3adc2ba367fb6
FlinkSQL>select*fromhadoop_catalog.iceberg_db.my_user
>/*+options(
>'streaming'='true',
>'monitor-interval'='30s',
>'start-snapshot-id'='138573494821828246'