SQLServer数据源

更新时间：2026-07-03

SQL Server 数据源支持离线库表采集（Jdbc-SQLserver）和实时库表采集（SqlServer-CDC）两类 source 插件，可写入 Iceberg、Doris、结构化数据集等目标端（各目标端插件参数详见各自文档）。本文以 Iceberg 作为目标端示例。

SQL Server 插件类型

插件	plugin_name	_db_pluginId	作业类型	说明
SQL Server 离线采集	`Jdbc`	`Jdbc-SQLserver`	`batch`	JDBC 读取 SQL Server 表，适用于离线全量/条件增量同步
SQL Server 准实时 CDC 采集	`SqlServer-CDC`	`SqlServer-CDC`	`streaming`	通过 SQL Server CDC 机制实现全量+增量或仅增量同步

脚本 Demo 与参数说明

一、离线任务：SQL Server → Iceberg

适用于离线全量同步或按条件增量同步（如按 updated_at 每日增量）。作业类型为 batch，source 使用 Jdbc-SQLserver，sink 使用 Iceberg-batch。

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "sqlserver_orders_to_iceberg_batch",
    "description": "SQL Server 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-SQLserver",
      "plugin_output": "source_table",
      "_db_connectionId": "sqlserver-conn-001",
      "_db_sourceDatabase": "dw",
      "_db_sourceSchema": "dbo",
      "_db_sourceTable": "orders",
      "_db_where": "updated_at >= ${logicTime(yyyy-MM-dd HH:mm:ss,-1d)}",
      "partition_column": "id",
      "_db_sourceChange": {
        "onDeleteColumn": "SKIP",
        "onAddColumn": "SKIP",
        "onDeleteSource": "PAUSE"
      }
    }
  ],
  "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": "订单表"
    }
  ]
}
            

Reader 参数：Jdbc-SQLserver

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `Jdbc`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `Jdbc-SQLserver`	是	无
`plugin_output`	Source 输出流名称，需与 sink 的 `plugin_input` 保持一致	是	无
`_db_connectionId`	SQL Server 数据源连接 ID	是	无
`_db_sourceDatabase`	数据库名称	是	无
`_db_sourceSchema`	Schema 名称（如 `dbo`）	是	无
`_db_sourceTable`	数据表名称	是	无
`partition_column`	分片字段，用于多并发采集，建议使用主键或唯一键	否	无
`_db_where`	数据过滤条件，不含 `WHERE` 关键字，支持时间宏如 `${logicTime(yyyy-MM-dd HH:mm:ss,-1d)}`	否	无
`_db_sourceChange.onDeleteColumn`	源端表字段被删除时的策略：`SKIP` 忽略 / `PAUSE` 终止任务	否	`SKIP`
`_db_sourceChange.onAddColumn`	源端表新增字段时的策略：`SKIP` 忽略 / `PAUSE` 终止任务	否	`SKIP`
`_db_sourceChange.onDeleteSource`	源端表被删除时的策略，当前仅支持 `PAUSE`	否	`PAUSE`

二、实时任务：SqlServer-CDC → Iceberg

适用于 SQL Server 变更数据准实时同步到 Iceberg 数据湖的场景。作业类型为 streaming，source 使用 SqlServer-CDC，sink 使用 Iceberg-streaming。示例为全量+增量模式：先进行全量快照，完成后自动切换为 CDC 增量消费。

                JSON
                
            

                {
  "_db_jobMeta": {
    "name": "sqlserver_orders_to_iceberg_streaming",
    "description": "SQL Server 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": "SqlServer-CDC",
      "_db_pluginId": "SqlServer-CDC",
      "plugin_output": "source_table",
      "_db_connectionId": "sqlserver-cdc-conn-001",
      "_db_sourceDatabase": "dw",
      "_db_sourceSchema": "dbo",
      "_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": {
        "sourceSchema": "dbo",
        "sourceTable": "orders",
        "sinkTable": "ods_orders"
      },
      "source": {
        "_db_sourceChange": {
          "onDeleteColumn": "SKIP",
          "onAddColumn": "SKIP",
          "onDeleteSource": "PAUSE",
          "onRenameColumn": "SKIP",
          "onTruncateTable": "SKIP",
          "onRenameTable": "SKIP",
          "onChangeColumn": "SKIP",
          "onChangeTableComment": "SKIP",
          "onChangeColumnComment": "SKIP"
        }
      },
      "sink": {
        "_db_isAutoCreated": true,
        "_db_sinkNameRule": "SAME",
        "_db_dmlConfig": {
          "insert": "INSERT",
          "update": "UPDATE",
          "delete": "DELETE",
          "logicalDeleteTag": ""
        }
      }
    }
  ]
}
            

Reader 参数：SqlServer-CDC

source[0] 全局参数

参数	描述	是否必选	默认值
`plugin_name`	SeaTunnel 插件名，固定填 `SqlServer-CDC`	是	无
`_db_pluginId`	DataBuilder 侧插件 ID，固定填 `SqlServer-CDC`	是	无
`plugin_output`	Source 输出流名称，需与 sink 的 `plugin_input` 保持一致	是	无
`_db_connectionId`	SQL Server CDC 数据源连接 ID	是	无
`_db_sourceDatabase`	数据库名称	是	无
`_db_sourceSchema`	Schema 名称（如 `dbo`）	否	无
`_db_syncMode`	同步类型：`BASE_INCREMENT`（全量+增量）/ `INCREMENT`（仅增量）	否	`BASE_INCREMENT`
`_db_incrementPosition.position`	指定起始 CDC 时间位点，需确保对应变更日志未被清理	否	无

表级 DDL 变更策略（写入 _db_tableConfigs[i].source._db_sourceChange）

参数	描述	可选值	默认值
`onDeleteColumn`	源端表字段被删除时的策略	`SKIP` / `PAUSE`	`PAUSE`
`onAddColumn`	源端表新增字段时的策略	`SKIP` / `PAUSE`	`SKIP`
`onDeleteSource`	源端表被删除时的策略	`PAUSE`	`PAUSE`
`onRenameColumn`	源端表字段被重命名时的策略	`SKIP` / `PAUSE`	`SKIP`
`onTruncateTable`	源端执行 `TRUNCATE TABLE` 时的策略	`SKIP` / `PAUSE`	`SKIP`
`onRenameTable`	源端表被重命名时的策略	`SKIP` / `PAUSE`	`SKIP`
`onChangeColumn`	源端字段类型变更时的策略	`SKIP` / `PAUSE` / `SYNC`	`SKIP`
`onChangeTableComment`	源端表描述变更时的策略	`SKIP` / `PAUSE` / `SYNC`	`SKIP`
`onChangeColumnComment`	源端字段描述变更时的策略	`SKIP` / `PAUSE` / `SYNC`	`SKIP`

配置编写注意事项

plugin_name 与 _db_pluginId 的区别：plugin_name 是引擎层的真实插件名，_db_pluginId 是 DataBuilder 平台侧标识。SQL Server 离线插件的 plugin_name = Jdbc，_db_pluginId = Jdbc-SQLserver；准实时插件两者均为 SqlServer-CDC。
离线 vs 实时的选型原则：
- 离线全量/条件过滤增量 → 作业类型 batch，source 用 Jdbc-SQLserver
- 准实时 CDC 同步 → 作业类型 streaming，source 用 SqlServer-CDC
SQL Server CDC 前提：使用 SqlServer-CDC 时，SQL Server 必须已在实例级和表级启用 CDC（sys.sp_cdc_enable_db + sys.sp_cdc_enable_table），且消费起点对应的变更日志未被清理。
全局配置 vs 表级配置：
- 离线（batch）：属性写在 source[0] / sink[0] 中
- 准实时（streaming）：表级策略（DDL 变更策略、DML 写入策略）写在 _db_tableConfigs[i].source / _db_tableConfigs[i].sink 中

评价此篇文章

有帮助没帮助

PostgreSQL数据源

DM数据源

百度智能云

百度胜算