文档首页/ 智能数据湖 AI DataLake/ 开发指南/ Iceberg开发指南/ Iceberg on Flink/ 查询与写入Iceberg
更新时间：2026-10-09 GMT+08:00
查看PDF
分享

查询与写入Iceberg

注意事项

  • 流式写入Iceberg时需要开启checkpoint。
  • Upsert写入的Iceberg表，不支持流读。

代码示例

示例1：无主键数据追加写入Iceberg表。

CREATE CATALOG lf_iceberg_catalog
WITH
  (
    'type' = 'iceberg',
    'catalog-type' = 'lakeformation'
  );
CREATE TABLE IF NOT EXISTS lf_iceberg_catalog.iceberg_db.iceberg_table (
  id BIGINT, 
  data STRING
)WITH('format-version' = '2');

CREATE TABLE kafka_source (id BIGINT, data STRING)
WITH
  (
    'connector' = 'kafka',
    'topic' = 'user-events',
    'properties.bootstrap.servers' = '127.0.0.1:9092',
    'properties.group.id' = 'flink-consumer-group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
  );

INSERT INTO
  lf_iceberg_catalog.iceberg_db.iceberg_table
SELECT
  *
FROM
  kafka_source

示例2：流读Iceberg表。

CREATE CATALOG lf_iceberg_catalog
WITH
  (
    'type' = 'iceberg',
    'catalog-type' = 'lakeformation'
  );

CREATE TABLE sink (id BIGINT, data STRING)
WITH
  ('connector' = 'print');

INSERT INTO
  sink
SELECT
  *
FROM
  lf_iceberg_catalog.iceberg_db.iceberg_table /*+ OPTIONS('streaming'='true', 'monitor-interval'='60s') */;
父主题： Iceberg on Flink

相关文档