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

数据湖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"}

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