SeaTunnel JDBC DuckDB Source Connector 完全指南：从本地数据库文件到分布式数据管道的读取实战

SeaTunnel JDBC DuckDB Source Connector 完全指南：从本地数据库文件到分布式数据管道的读取实战

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

SeaTunnel 通过 JDBC Source Connector 支持读取 DuckDB 数据库中的数据。DuckDB 是一款进程内（in-process）SQL 分析型（OLAP）数据库，没有独立的远程服务端，连接器直接面向本地数据库文件或内存数据库工作。本指南将完整讲解该连接器的能力边界、依赖安装、全部 Source 配置参数、数据类型映射、并行切片读取原理，并结合仓库源码给出可直接复制运行的实战配置。

概述与适用场景

DuckDB 以单文件数据库的形式存在，通常用于本地分析、数据探索与小型数据管道。SeaTunnel 的 JDBC DuckDB Source 通过jdbc:duckdb:/path/to/database.db这样的 JDBC URL 直接打开本地数据库文件，也可以连接内存数据库，例如jdbc:duckdb:memory:。

从连接器角度看，DuckDB 场景有以下几个显著特点：

• 无远程服务器：连接串指向本地文件路径或memory:，不存在 host/port 概念；
• 批量读取友好：官方特性表中batch、exactly-once、column projection、parallelism、user-defined split均为支持状态，stream流式模式不适用（嵌入式数据库本身没有持续变更日志可供消费）；
• 查询即投影：连接器支持自定义查询 SQL，通过select指定列即可实现列裁剪（projection）效果；
• 多表一次作业读取：通过table_list可以在一个 Source 中并行读取同一数据库文件内的多张表。

该连接器的官方文档位于 docs/en/connectors/source/DuckDB.md，底层实现归属于 JDBC 连接器模块，相关源码集中在seatunnel-connectors-v2/connector-jdbc。

支持版本与运行引擎

提示：不同依赖版本的 DuckDB 驱动类名可能不同，配置前请先确认当前duckdb_jdbc驱动包版本对应的 Driver 类。官方文档给出的驱动类为org.duckdb.DuckDBDriver。

使用依赖：驱动包放置位置

DuckDB 驱动 jar（duckdb_jdbc，可通过 Maven 中央仓库获取）需要手动放置到 SeaTunnel 安装目录中，放置位置取决于运行引擎：

• Spark / Flink 引擎：将驱动 jar 放入${SEATUNNEL_HOME}/plugins/目录；
• SeaTunnel Zeta 引擎：将驱动 jar 放入${SEATUNNEL_HOME}/lib/目录。

放置完成后即可在作业配置中使用Jdbc插件声明 DuckDB 数据源。

连接器能力清单

以下是官方特性清单（详见 connector-v2-features）：

其中"列投影"能力的实现方式为：支持自定义查询 SQL，通过查询语句控制读取的字段集合。

数据源信息速查

从源码看，JDBC 连接器通过 DuckDBDialectFactory 中的acceptsURL方法以url.startsWith("jdbc:duckdb:")识别 DuckDB 方言，因此 URL 前缀jdbc:duckdb:是判别该数据源类型的硬性条件。

Source 配置参数详解

连接器在作业配置中的插件名为Jdbc（JDBC 连接器统一插件名）。完整参数如下：

参数实现细节与取值建议

从源码 JdbcSourceOptions 中可以确认以下实现细节：

• split.size默认值 8096（定义于JdbcSourceOptions.java#L49-L54），即每 8096 行切一个 split；
• fetch_size默认 0，表示交由 JDBC 驱动使用默认抓取大小；
• table_path、where_condition、table_list均为无默认值的可选参数，其中table_list被定义为List<JdbcSourceTableConfig>结构类型；
• 除文档列出的参数外，JDBC 源还提供一组split.*高级调优参数，例如：
  • split.even-distribution.factor.upper-bound（默认 100.0）与split.even-distribution.factor.lower-bound（默认 0.05）：用于判定表数据分布是否均匀，分布因子计算公式为(MAX(id) - MIN(id) + 1) / rowCount，均匀分布时走均匀切分优化，不均匀时走查询式切分；
  • split.sample-sharding.threshold（默认 1000）、split.inverse-sampling.rate（默认 1000）、split.allow-sampling（默认 true）：控制大数据量下基于采样的分片策略；
  • use_select_count（默认 false）、skip_analyze（默认 false）：控制表行数的统计方式。

这些参数同样适用于 DuckDB 数据源，可在需要精细控制切片行为时使用。

关于 table_list 的结构约束

table_list中每一项都对应一张表，从 JdbcSourceTableConfig 源码可见，每项可独立配置：

• table_path：表完整路径（必填项）；
• query：该表自定义查询，用于过滤行与列；
• partition_column/partition_num/partition_lower_bound/partition_upper_bound：该表的独立并行切分配置；
• use_select_count/skip_analyze/use_regex：该表的统计与正则匹配开关。

需要注意两点实现约束：

1. 当table_list中配置了多张表时，各表的table_path必须唯一，不允许为空或重复，否则校验会直接抛异常（见JdbcSourceTableConfig.java#L108-L119）；
2. 当表项未显式指定partition_num时，会使用默认值10（见JdbcSourceTableConfig.java#L42）。

数据类型映射

DuckDB 类型到 SeaTunnel 类型的官方映射关系如下：

源码层面的类型转换细节

类型映射的落地实现位于 DuckDBTypeConverter，并通过 DuckDBTypeMapper 接入 JDBC 方言体系。从当前仓库源码看，实际处理比文档表格更细：

• DECIMAL 精度上限 38、默认精度 18、最大小数位 38、默认小数位 3（见DuckDBTypeConverter.java#L80-L83）。当精度或小数位超限时会做截断并输出 warning 日志；小数位为负时归 0；
• TIMESTAMP WITH TIME ZONE 映射为 OFFSET_DATE_TIME 类型（见DuckDBTypeConverter.java#L169-L171），即带时区偏移的时间类型，而非普通 TIMESTAMP。该行为在 DuckDBTypeConverterTest 中有明确断言；
• 复杂类型 ARRAY / STRUCT / MAP 实际映射为 STRING（默认长度 65535），转换时会输出 warning 日志提示"复杂类型已映射为 STRING，可考虑使用 JSON 序列化"（见DuckDBTypeConverter.java#L176-L184），而不是 BYTES；BLOB 才映射为 BYTES（PrimitiveByteArrayType）；
• 无符号整数族（UTINYINT、USMALLINT、UINTEGER、UBIGINT）会分别落到对应的有符号 SeaTunnel 类型；HUGEINT / UHUGEINT / BIGNUM 统一映射为DECIMAL(38,0)；
• 遇到未知类型（如geography）时会回退为 STRING 并输出 warning（DuckDBTypeConverter.java#L185-L189）。

上述行为均以当前仓库源码为准，若你使用的 SeaTunnel 发行版本不同，映射结果可能略有差异，建议以对应版本源码为准。

并行读取（Parallel Reader）原理

JDBC Source 支持对表数据进行并行读取。SeaTunnel 会按一定规则将表数据切分为多个 split，再交给多个 reader 并行消费，reader 数量由作业的parallelism决定。

Split 键选择规则

1. 显式指定优先：若配置了partition_column，直接使用该列计算 split，该列必须属于"受支持的 split 数据类型"；
2. 自动推导兜底：若未配置partition_column，SeaTunnel 会读取表 schema 获取主键（Primary Key）与唯一索引（Unique Index）。当主键/唯一索引包含多列时，取其中第一个属于受支持 split 数据类型的列用于切分。例如表主键为(guid, name varchar)，由于guid不属于受支持类型，会自动改用name列切分。

受支持的 split 数据类型

• String（字符串）
• Number（数值类型：int、bigint、decimal 等）
• Date（日期）

与切片相关的参数

注意：partition_num在官方文档中标注为"不推荐使用"，因为通过split.size控制每片行数更能适配数据量的动态变化。

无法切分时的行为

如果表既没有主键/唯一索引，也未设置partition_column，则该表以单并发（single concurrency）方式运行。对于这种场景，可开启enable_concurrent_read = false让源在快照阶段跳过切片分析、按单个 split 读取，适合无索引的大表（该选项定义于 JdbcSourceOptions）。

单表读取与多表读取的配置选择

官方 Tips 给出两条核心建议：

1. 单表读取：优先用table_path替代query。配置table_path会自动开启自动切片（auto split），可通过split.*参数调整切片策略；
2. 多表读取：使用table_list。配置table_list同样会自动开启自动切片。

从 DuckDBDialect 源码可见，DuckDB 方言将默认数据库名设为default、默认 schema 设为main，表标识符统一用双引号包裹（如"main"."table1"）。因此table_path写作main.user_events等价于"main"."user_events"。同时该方言刻意不支持 UPSERT 语义（getUpsertStatement返回空Optional），这是出于批量 ETL 负载与追加写入优化的设计取舍。

实战任务示例

以下示例均以本地 DuckDB 数据库文件/tmp/test.db为数据源，Sink 使用 Console 插件将结果输出到控制台，可直接替换数据库路径与表名后运行。

示例一：单并行简单查询

该示例以单并行度查询测试库中的user_events表并输出全部字段；你也可以通过修改query指定要查询的字段，实现列裁剪。

# Defining the runtime environment env { parallelism = 4 job.mode = "BATCH" } source{ Jdbc { url = "jdbc:duckdb:/tmp/test.db" driver = "org.duckdb.DuckDBDriver" connection_check_timeout_sec = 100 username = "duckdb" password = "" query = "select * from user_events limit 16" } } transform { # 如需了解 transform 插件的更多配置，请参考 SeaTunnel 官方 transform 文档 } sink { Console {} }

示例二：按 partition_column 并行读取

通过partition_column指定切分列，配合split.size控制每片行数。上下边界可通过partition_lower_bound/partition_upper_bound显式指定（注释掉则自动获取）。

env { parallelism = 4 job.mode = "BATCH" } source { Jdbc { url = "jdbc:duckdb:/tmp/test.db" driver = "org.duckdb.DuckDBDriver" connection_check_timeout_sec = 100 username = "duckdb" password = "" query = "select * from user_events" partition_column = "id" split.size = 10000 # Read start boundary #partition_lower_bound = ... # Read end boundary #partition_upper_bound = ... } } sink { Console {} }

示例三：按主键或唯一索引自动并行

配置table_path会开启自动切片，读取时 SeaTunnel 会自动从表的主键/唯一索引中挑选合适的切分列。示例中同时保留了query，用于在自动切片基础上限定读取范围。

env { parallelism = 4 job.mode = "BATCH" } source { Jdbc { url = "jdbc:duckdb:/tmp/test.db" driver = "org.duckdb.DuckDBDriver" connection_check_timeout_sec = 100 username = "" password = "" table_path = "main.user_events" query = "select * from main.user_events" split.size = 10000 } } sink { Console {} }

示例四：指定并行边界

显式声明partition_lower_bound与partition_upper_bound后，只读取上下边界范围内的数据，相比全表扫描更高效。同时可通过properties向 DuckDB 传递额外连接参数，例如线程数与内存上限。

source { Jdbc { url = "jdbc:duckdb:/tmp/test.db" driver = "org.duckdb.DuckDBDriver" connection_check_timeout_sec = 100 username = "duckdb" password = "" # Define query logic as required query = "select * from user_events" partition_column = "id" # Read start boundary partition_lower_bound = 1 # Read end boundary partition_upper_bound = 500 partition_num = 10 properties { threads=4 memory_limit="4GB" } } }

说明：DuckDB 允许在 JDBC URL 或properties中传递连接参数（如threads、memory_limit）。官方文档指出，当 properties 与 URL 参数同名时，DuckDB 中 properties 优先于 URL。memory_limit用于限制 DuckDB 分析引擎可使用的内存上限，threads控制其内部并行线程数，合理设置有助于在受限环境中稳定运行。

示例五：多表读取

通过table_list一次读取多张表。每张表可以独立配置query实现行/列过滤；未配置query的表默认读取全量字段。注释部分展示了where_condition（全局过滤）与split.size（切片行数）的用法。

env { job.mode = "BATCH" parallelism = 4 } source { Jdbc { url = "jdbc:duckdb:/tmp/test.db" driver = "org.duckdb.DuckDBDriver" connection_check_timeout_sec = 100 username = "duckdb" password = "" table_list = [ { table_path = "main.table1" }, { table_path = "main.table2" # Use query filter rows & columns query = "select id, name from main.table2 where id > 100" } ] #where_condition= "where id > 100" #split.size = 8096 } } sink { Console {} }

源码级原理补充：方言、URL 解析与 Catalog

为了更透彻地理解该连接器，可以关注以下四个源码切入点：

1. 方言识别与注册：DuckDBDialectFactory 通过@AutoService(JdbcDialectFactory.class)自动注册，acceptsURL以 URL 前缀jdbc:duckdb:判定归属；DuckDBDialect 负责表路径解析、标识符引用（双引号）与类型映射器的装配；
2. URL 解析：DuckDBURLParser 使用正则^jdbc:duckdb:(?<path>[^?]*?)(?<suffix>\?.*)?$提取数据库文件路径与查询参数后缀，天然兼容jdbc:duckdb:memory:内存库形式，host/port 记为 localhost/0；
3. Catalog 支持：DuckDBCatalogFactory 提供表结构推断能力（可选参数包括schema、decimal_type_narrowing、handle_blob_as_string），默认 schema 为main，这为table_path/table_list自动切片时的 schema 推断提供了基础；
4. 端到端验证：仓库内的 DuckDBSourceAndSinkTest 展示了覆盖 BOOLEAN、TINYINT、HUGEINT、无符号整数族、REAL、DECIMAL、VARCHAR/TEXT/CHAR/BPCHAR、BLOB、DATE/TIME/TIMESTAMP/TIMESTAMPTZ、INTERVAL、UUID 等全部类型的一张建表语句，并通过真实 JDBC 连接完成 Source 读取与 Sink 写入的完整流程验证；DuckDBConnectDryRunValidationTest 则验证了--dry-run connect钩子下的 schema 推断与连通性检查。

如果你需要在批处理作业中快速验证连接配置，可以参考仓库根目录下的 config/v2.batch.config.template 模板，将其中 source 部分替换为上述 DuckDB 配置即可。

常见注意点汇总

• 无法切分则单并发：表无主键/唯一索引且未设置partition_column时，作业以单并发运行，吞吐会受限，建议为表补充主键或显式配置切分列；
• 优先table_path/table_list：单表用table_path，多表用table_list，二者均自动开启切片；query更灵活但需要自己控制切分边界；
• 驱动放置位置随引擎变化：Spark/Flink 放plugins/，Zeta 放lib/，放错目录会报 ClassNotFound；
• 版本差异：DuckDB 不同版本驱动类名可能不同，务必核对实际驱动包；
• 类型映射以源码为准：当前仓库中 TIMESTAMP WITH TIME ZONE 映射为带时区偏移的时间类型、ARRAY/STRUCT/MAP 映射为 STRING，若与文档表格不一致，以发行版本对应源码为准；
• 表路径唯一性：table_list多表场景下各表table_path必须唯一，否则作业校验失败。

总结

SeaTunnel JDBC DuckDB Source 是打通"本地嵌入式 OLAP 分析库"与"分布式数据管道"的轻量桥梁：它没有网络服务依赖，只需驱动 jar 与文件路径即可接入；通过partition_column、主键/唯一索引自动切片与split.size等机制可以获得稳定的并行读取能力；table_path/table_list让单表与多表批量读取的配置成本都保持在极低水平。结合本指南中的参数表、类型映射与五个可直接运行的示例，你可以快速将 DuckDB 中的数据导入 SeaTunnel 支持的任意下游存储。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel