DB-结构化数据集

更新时间：2026-07-03

结构化数据集目标端支持离线批量写入（StructuredDataSet-batch）和实时流式写入（StructuredDataSet-streaming）两类 sink 插件，用于将源端库表数据写入项目内结构化数据集。本文以 MySQL 作为源端示例。

结构化数据集插件类型

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

脚本 Demo 与参数说明

一、离线任务：MySQL → 结构化数据集

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

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "mysql_to_structured_dataset_batch",
    "description": "MySQL orders 表离线同步到结构化数据集",
    "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": "StructuredDataSet",
      "_db_pluginId": "StructuredDataSet-batch",
      "plugin_input": "source_table",
      "_db_sinkParentFolderId": "folder-001",
      "iceberg.table.upsert-mode-enabled": false,
      "data_save_mode": "APPEND_DATA",
      "_db_isAutoCreated": true,
      "_db_sinkNameRule": "SAME",
      "_db_comment": "订单数据集"
    }
  ]
}
            

Writer 参数：StructuredDataSet-batch

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `StructuredDataSet`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `StructuredDataSet-batch`	是	无
`plugin_input`	对应 source 的 `plugin_output` 值	是	无
`_db_sinkParentFolderId`	目标项目文件夹 ID	是	无
`iceberg.table.upsert-mode-enabled`	是否启用 Upsert 写入模式（基于主键去重更新）	否	`false`
`data_save_mode`	写入模式：`APPEND_DATA`（追加）/ `DROP_DATA`（覆盖）	否	`APPEND_DATA`
`_db_isAutoCreated`	是否自动创建结构化数据集：`true`（自动创建）/ `false`（选择已有数据集）	否	`true`
`_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_sinkTable`	目标数据集名称或选择已有目标数据集	否	无
`_db_comment`	目标数据集描述（可选）	否	无

二、实时任务：MySQL-CDC → 结构化数据集

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

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "mysql_cdc_to_structured_dataset_streaming",
    "description": "MySQL orders 表 CDC 同步到结构化数据集",
    "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": "StructuredDataSet",
      "_db_pluginId": "StructuredDataSet-streaming",
      "plugin_input": "source_table",
      "_db_sinkParentFolderId": "folder-001"
    }
  ],
  "_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 参数：StructuredDataSet-streaming

sink[0] 全局参数

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `StructuredDataSet`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `StructuredDataSet-streaming`	是	无
`plugin_input`	对应 source 的 `plugin_output` 值	是	无
`_db_sinkParentFolderId`	目标项目文件夹 ID	是	无
`_db_structuredDataFileId`	已有结构化数据集 FileId，选择已有数据集时使用	否	无

表级配置（写入 _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_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`

配置编写注意事项

离线 vs 流式的选型原则：
- 离线全量/条件过滤增量 → 作业类型 batch，sink 用 StructuredDataSet-batch
- CDC 准实时同步 → 作业类型 streaming，sink 用 StructuredDataSet-streaming
Upsert 与覆盖写入：StructuredDataSet-batch 支持 iceberg.table.upsert-mode-enabled 和 data_save_mode；覆盖写入需谨慎使用。
Streaming 表级配置：流式写入的建表方式、命名规则、DML 策略均在 _db_tableConfigs[i].sink 中配置，不在 sink[0] 中配置。

评价此篇文章

有帮助没帮助

DB-doris

开发机

百度智能云

百度胜算