DB-iceberg

更新时间：2026-07-03

Iceberg 数据源支持离线库表写入（Iceberg-batch）和实时库表写入（Iceberg-streaming）两类 sink 插件，是 DataBuilder 数据湖仓的核心写入目标。本文以 MySQL 作为源端示例。

DB-Iceberg 插件类型

插件	plugin_name	_db_pluginId	作业类型	说明
Iceberg 离线写入	`Iceberg`	`Iceberg-batch`	`batch`	离线批量写入 Iceberg 表，支持追加/覆盖写入和 Upsert 模式
Iceberg 流式写入	`Iceberg`	`Iceberg-streaming`	`streaming`	准实时 CDC 变更数据写入 Iceberg 表，支持 INSERT/UPDATE/DELETE 事件处理

脚本 Demo 与参数说明

一、离线任务：MySQL →DB-Iceberg

作业类型为 batch，source 使用 Jdbc-Mysql（参数详见 MySQL 数据源文档），sink 使用 Iceberg-batch。

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "mysql_to_iceberg_batch",
    "description": "MySQL orders 表离线同步到 Iceberg",
    "parentFolderId": "project_91cc_799bd26eee94",
    "type": "batch",
    "mode": "script"
  },
  "env": {
    "job.mode": "BATCH",
    "parallelism": 1,
    "checkpoint.interval": 30000
  },
  "source": [
    {
      "plugin_name": "Jdbc",
      "_db_pluginId": "Jdbc-Mysql",
      "plugin_output": "source_table",
      "_db_connectionId": "mysql-conn-001",
      "_db_sourceDatabase": "sales_db",
      "_db_sourceTable": "orders"
    }
  ],
  "sink": [
    {
      "plugin_name": "Iceberg",
      "_db_pluginId": "Iceberg-batch",
      "plugin_input": "source_table",
      "_db_sinkPath": "lake.default",
      "_db_catalog": "lake",
      "_db_schema": "default",
      "iceberg.table.upsert-mode-enabled": false,
      "data_save_mode": "APPEND_DATA",
      "_db_isAutoCreated": true,
      "_db_sinkTableType": "MANAGED",
      "_db_sinkNameRule": "SAME",
      "_db_comment": "订单表"
    }
  ]
}
            

Writer 参数：Iceberg-batch

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `Iceberg`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `Iceberg-batch`	是	无
`plugin_input`	对应 source 的 `plugin_output` 值	是	无
`_db_sinkPath`	目标 Catalog 和 Schema，格式为 `{catalog}.{schema}`，如 `lake.default`	是	无
`_db_catalog`	目标 Catalog 名称	是	无
`_db_schema`	目标 Schema 名称	是	无
`iceberg.table.upsert-mode-enabled`	是否启用 Upsert 写入模式（基于主键去重更新）	否	`false`
`data_save_mode`	写入模式：`APPEND_DATA`（追加）/ `DROP_DATA`（覆盖）	否	`APPEND_DATA`
`_db_isAutoCreated`	是否自动建表：`true`（自动建表）/ `false`（选择已有表）	否	`true`
`_db_sinkTableType`	表类型，当前固定为 `MANAGED`（托管表）	否	`MANAGED`
`_db_sinkNameRule`	目标表命名规则：`SAME`（与源表同名）/ `ADD_PREFIX` / `ADD_SUFFIX` / `ADD_PREFIX_AND_SUFFIX` / `CUSTOM`	否	`SAME`
`_db_prefix`	表名前缀，`_db_sinkNameRule` 含 `ADD_PREFIX` 时生效	否	无
`_db_suffix`	表名后缀，`_db_sinkNameRule` 含 `ADD_SUFFIX` 时生效	否	无
`_db_sinkTableName`	自定义表名，`_db_sinkNameRule=CUSTOM` 时使用	否	无
`_db_sinkTable`	选择已有目标表，`_db_isAutoCreated=false` 时使用	否	无
`_db_comment`	目标表描述（可选）	否	无

二、实时任务：MySQL-CDC → DB-Iceberg

作业类型为 streaming，source 使用 MySQL-CDC（参数详见 MySQL 数据源文档），sink 使用 Iceberg-streaming。

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "mysql_cdc_to_iceberg_streaming",
    "description": "MySQL orders 表 CDC 同步到 Iceberg",
    "parentFolderId": "folder_af42_bccdb16a1d00",
    "type": "streaming",
    "mode": "script",
    "syncMode": "BASE_INCREMENT"
  },
  "env": {
    "job.mode": "STREAMING",
    "parallelism": 1,
    "_db_enableCheckpoint": false
  },
  "source": [
    {
      "plugin_name": "MySQL-CDC",
      "_db_pluginId": "MySQL-CDC",
      "plugin_output": "source_table",
      "_db_connectionId": "mysql-cdc-conn-001",
      "_db_sourceDatabase": "sales_db",
      "_db_syncMode": "BASE_INCREMENT"
    }
  ],
  "sink": [
    {
      "plugin_name": "Iceberg",
      "_db_pluginId": "Iceberg-streaming",
      "plugin_input": "source_table",
      "_db_sinkPath": "lake.default",
      "_db_catalog": "lake",
      "_db_schema": "default",
      "_db_sinkTableType": "MANAGED"
    }
  ],
  "_db_tableConfigs": [
    {
      "tableIdentity": {
        "sourceTable": "orders",
        "sinkTable": "ods_orders"
      },
      "sink": {
        "_db_isAutoCreated": true,
        "_db_sinkNameRule": "SAME",
        "_db_dmlConfig": {
          "insert": "INSERT",
          "update": "UPDATE",
          "delete": "DELETE",
          "logicalDeleteTag": ""
        }
      }
    }
  ]
}
            

Writer 参数：Iceberg-streaming

sink[0] 全局参数

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `Iceberg`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `Iceberg-streaming`	是	无
`plugin_input`	对应 source 的 `plugin_output` 值	是	无
`_db_sinkPath`	目标 Catalog 和 Schema，格式为 `{catalog}.{schema}`，如 `lake.default`	是	无
`_db_catalog`	目标 Catalog 名称	是	无
`_db_schema`	目标 Schema 名称	是	无
`_db_sinkTableType`	表类型，当前固定为 `MANAGED`（托管表）	否	`MANAGED`

表级配置（写入 _db_tableConfigs[i].sink）

参数	描述	可选值	默认值
`_db_isAutoCreated`	是否自动建表	`true` / `false`	`true`
`_db_sinkNameRule`	目标表命名规则	`SAME` / `ADD_PREFIX` / `ADD_SUFFIX` / `ADD_PREFIX_AND_SUFFIX` / `CUSTOM`	`SAME`
`_db_prefix`	表名前缀	任意字符串	无
`_db_suffix`	表名后缀	任意字符串	无
`_db_sinkTable`	选择已有目标表，`_db_isAutoCreated=false` 时使用	任意字符串	无
`_db_comment`	目标表描述（可选）	任意字符串	无
`_db_dmlConfig.insert`	INSERT 事件处理策略	`INSERT` / `IGNORE`	`INSERT`
`_db_dmlConfig.update`	UPDATE 事件处理策略	`UPDATE` / `IGNORE`	`UPDATE`
`_db_dmlConfig.delete`	DELETE 事件处理策略；逻辑删除会写入 `logical_delete_tag` 字段	`DELETE` / `IGNORE` / `LOGICAL_DELETE`	`DELETE`

配置编写注意事项

plugin_name 相同，_db_pluginId 不同：Iceberg-batch 和 Iceberg-streaming 的 plugin_name 均为 Iceberg，通过 _db_pluginId 区分。
离线 vs 流式的选型原则：
- 离线全量/条件过滤增量 → 作业类型 batch，sink 用 Iceberg-batch
- CDC 实时同步 → 作业类型 streaming，sink 用 Iceberg-streaming
Upsert 模式：Iceberg-batch 支持 iceberg.table.upsert-mode-enabled=true，在源端数据有重复主键时执行插入或更新，避免重复写入。
覆盖写入：data_save_mode=DROP_DATA 会先清空目标表再写入，适用于全量覆盖场景；日常增量同步使用 APPEND_DATA。
Streaming 表级配置：流式写入的建表方式、命名规则、DML 策略均在 _db_tableConfigs[i].sink 中配置，不在 sink[0] 中配置。

评价此篇文章

有帮助没帮助

DB-数据卷

DB-媒体集

百度智能云

百度胜算