数据湖Iceberg-FlinkSQL-kafka类型表数据无法成功写入(6)

news/2024/10/26 13:22:32/

数据湖Iceberg-简介(1)
数据湖Iceberg-存储结构(2)
数据湖Iceberg-Hive集成Iceberg(3)
数据湖Iceberg-SparkSQL集成(4)
数据湖Iceberg-FlinkSQL集成(5)
数据湖Iceberg-FlinkSQL-kafka类型表数据无法成功写入(6)
数据湖Iceberg-Flink DataFrame集成(7)

数据湖Iceberg-FlinkSQL-kafka类型表数据无法成功写入

数据湖Iceberg-FlinkSQL-kafka类型表数据无法成功写入版本问题问题原因解决方法

版本

Iceberg：1.1.0

Flink：1.14.3

问题

Kafka类型的Iceberg表创建完成后，通过语句写入其他表中执行成功，但是没数据

问题原因

当前版本的BUG（存疑）

解决方法

Kafka表必须要在default_catalog.default_database下，即catalog名为default_catalog，数据库(命名空间)为default_database下,否则kafka类型的表读取不到数据。

如果都在我们自己创建的catalog下创建，则执行INSERT INTO hadoop_catalog.iceberg_db.sample6 SELECT * FROM default_catalog.default_database.kafka1;后,在Flink任务中看不到一个持续执行的Flink Job，而正常执行该命令Flink会执行一个持续执行的任务，去消费kafka数据写入Iceberg，正常情况如下图：

在这里插入图片描述

所以这里我们kafka表在default_catalog.default_database下，写入数据的表在我们自己创建的hadoop_catalog.iceberg_db下

create table default_catalog.default_database.kafka1(id int,data string
) with ('connector' = 'kafka','topic' = 'ttt','properties.zookeeper.connect' = '172.16.24.194:2181','properties.bootstrap.servers' = '172.16.24.194:9092','format' = 'json','properties.group.id'='iceberg1','scan.startup.mode'='earliest-offset'
);CREATE TABLE `hadoop_catalog`.`iceberg_db`.`sample6` (`id`  INT UNIQUE COMMENT 'unique id',`data` STRING NOT NULL,PRIMARY KEY(`id`) NOT ENFORCED
) with (
'format-version'='2', 
'write.upsert.enabled'='true'
);INSERT INTO hadoop_catalog.iceberg_db.sample6 SELECT * FROM default_catalog.default_database.kafka1;

此时我们往Kafka发送数据：

{"id":123,"data":"llalalala"}
{"id":1123,"data":"asdfasfds"}

查看表中数据可以看到写入成功

select * from hadoop_catalog.iceberg_db.sample6;

在这里插入图片描述

再次发送数据

{"id":123,"data":"JastData"}

查看表中数据,发现修改成功

在这里插入图片描述