Flink集成Iceberg

本文提供Flink集成Iceberg场景的实施方案，包括配置示例、执行步骤和参数说明。

案例

使用flink对Iceberg数据进行增删改查

解决方案

创建Flink SQL查询作业，下载依赖

• iceberg-flink-runtime-1.18-1.6.1.jar下载
• libthrift-0.9.3.jar下载
• libfb303-0.9.3.jar下载
• hive-exec-3.1.3.jar下载
• hive-Standalone-metastore-3.1.3.jar下载
• hive-common-3.1.3.jar下载
• hadoop-mapreduce-client-core-3.3.5.jar下载
• commons-lang-2.6.jar下载

Flink SQL

CREATE CATALOG hive_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://172.19.189.246:9083',
  'clients'='10',
  'property-version'='1',
  'warehouse'='hdfs://172.19.189.246:9000/user/hive/warehouse'
);

CREATE TABLE IF NOT EXISTS  `hive_catalog`.`ispong_db`.`iceberg_table2` (
    id BIGINT COMMENT 'unique id',
    data STRING
);

-- 删除。更新 需要走批处理模式
INSERT INTO `hive_catalog`.`ispong_db`.`iceberg_table2` VALUES (1, 'a');

INSERT INTO `hive_catalog`.`ispong_db`.`iceberg_table2` VALUES (2, 'b');