SeaTunnel Kingbase人大金仓JDBC Source 连接器配置详解、类型映射与源码级拆分策略解析【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文基于 SeaTunnel 仓库中 Kingbase 源连接器官方文档docs/en/connectors/source/Kingbase.md系统讲解如何通过 JDBC 方式读取人大金仓 KingbaseES 数据库数据涵盖驱动部署、完整 Source 配置参数、数据类型映射、并行分片partition与表路径table_path等实战配置并结合仓库中connector-jdbc模块的 Kingbase 方言实现源码深入剖析方言识别、类型转换 fallback 机制与大表切分策略帮助你既能快速配置一个可运行的 Kingbase 同步任务又能理解底层实现的选型依据。一、连接器能力概览Kingbase 是 SeaTunnel 通过 JDBC 方式支持的国产数据库连接器KingbaseES V8R68.6与 PostgreSQL 兼容因此该连接器在connector-jdbc模块内以独立方言Dialect形式实现。根据官方文档与源码其能力矩阵如下特性支持情况说明batch批处理支持以job.mode BATCH运行stream流式不支持纯 JDBC 全量读取无 CDC 能力exactly-once不支持批模式下无源端精确一次语义column projection列投影支持下游只取部分列时查询自动裁剪列parallelism并行度支持通过partition_*系列参数按分片并行读取support user-defined split自定义拆分支持可指定分片边界控制读取范围文档声明支持的引擎为Spark、Flink、SeaTunnel Zeta即该 Source 配置可在这三类引擎中运行。二、数据源信息与驱动部署2.1 支持的数据源版本Kingbase 连接器的接入要素如下Datasource支持版本Driver 类名JDBC URL 示例Kingbase8.6com.kingbase8.Driverjdbc:kingbase8://localhost:54321/db_test由于 Kingbase 官方驱动不在 SeaTunnel 发行包的依赖中需要手动下载驱动 jar 并拷贝到插件目录# 下载 kingbase8-8.6.0.jar 后执行 cp kingbase8-8.6.0.jar $SEATUNNEL_HOME/plugins/jdbc/lib/2.2 方言如何被识别源码证据从源码结构看JDBC 连接器通过 SPI 注册的方言工厂识别数据库。KingbaseDialectFactory.java 中的关键逻辑非常直接Override public boolean acceptsURL(String url) { return url.startsWith(jdbc:kingbase8:); } Override public JdbcDialect create() { return new KingbaseDialect(); }也就是说只要url以jdbc:kingbase8:开头SeaTunnel 就会自动选用 Kingbase 方言类型映射、标识符引用规则、行转换器等全部走 Kingbase 专属实现。这解释了为什么driver必须配置为com.kingbase8.Driver而 URL 前缀不可写成jdbc:postgresql:——后者会命中 PostgreSQL 方言。三、数据类型映射3.1 官方映射表Kingbase 到 SeaTunnel 的类型映射如下摘自官方文档Kingbase 数据类型SeaTunnel 数据类型BOOLBOOLEANINT2SHORTSMALLSERIAL / SERIAL / INT4INTINT8 / BIGSERIALBIGINTFLOAT4FLOATFLOAT8DOUBLENUMERICDECIMAL(precision, scale)取列定义的小数位数BPCHAR / CHARACTER / VARCHAR / TEXTSTRINGTIMESTAMPLOCALDATETIMETIMELOCALTIMEDATELOCALDATE其他类型暂不支持3.2 源码级深挖fallback 类型转换上表是文档口径的基础映射而仓库源码中的 KingbaseTypeConverter.java 揭示了更完整的实际行为AutoService(TypeConverter.class) public class KingbaseTypeConverter extends PostgresTypeConverter {Kingbase 类型转换器继承自PostgresTypeConverter印证了 KingbaseES 对 PostgreSQL 的兼容性并在此之上做三层扩展Kingbase 独有类型的兜底处理convert()先调用父类PostgresTypeConverter转换父类抛异常时进入 switch 兜底分支支持TINYINT→BYTE_TYPEPG 无此类型Kingbase 兼容 MySQL 语法提供MONEY→DECIMAL(38, 18)BLOB→ 二进制类型PrimitiveByteArrayType长度上限 1GBCLOB→STRING长度上限 1GBBIT(M)→ 二进制类型按M/8向上取整换算成字节长度。跨库兼容类型由于 KingbaseES 具备 MySQL/Oracle 兼容模式转换器还纳入了INT、MEDIUMINT、DATETIME、BLOB/TEXT系列MySQL 风格、NUMBER、VARCHAR2、NVARCHAR2、XMLOracle 风格、DATETIME2、DATETIMEOFFSETSQL Server 风格等类型名避免兼容模式建表时报不支持的类型错误。超限时降级而非报错例如DATETIME的 scale 超过MAX_TIMESTAMP_SCALE时会截断到最大 scale 并打 warn 日志而不中断任务。读取路径上KingbaseTypeMapper.java 从ResultSetMetaData中取列名、getColumnTypeName()原始类型、精度与小数位组装成BasicTypeDefine后交给KingbaseTypeConverter完成最终映射——这就是文档中NUMERIC能自动映射出精确DECIMAL(precision, scale)的实现来源。行数据的转换则由 KingbaseJdbcRowConverter.java 负责与 PostgreSQL 的 JDBC 行转换逻辑保持一致。四、Source Options 完整参数表以下参数与官方文档完全对齐Kingbase 与 PostgreSQL 等其他 JDBC 方言共用这套 Source 选项定义见 JdbcSourceOptions.javaNameType必填默认值描述urlString是-JDBC 连接 URL示例jdbc:kingbase8://localhost:54321/testdriverString是-连接远程数据源的 JDBC 类名应为com.kingbase8.DriverusernameString否-连接实例用户名旧配置键user仍作为 fallback 被接受passwordString否-连接实例密码queryString是-查询语句connection_check_timeout_secInt否30校验连接可用性时等待数据库操作完成的秒数partition_columnString否-并行分片列仅支持数值类型和字符串类型列partition_lower_boundBigDecimal否-分片列最小扫描值不设置时 SeaTunnel 会查询数据库获取 min 值partition_upper_boundBigDecimal否-分片列最大扫描值不设置时 SeaTunnel 会查询数据库获取 max 值partition_numInt否作业并行度分片数量仅支持正整数默认为作业并行度fetch_sizeInt否0大结果集查询的行抓取大小减少数据库往返次数以提升性能0 表示使用 JDBC 默认值use_regexBoolean否false控制table_path是否按正则匹配。为true时按正则模式匹配否则按精确路径table_pathString否-表全路径可替代query例如test_schema.table1table_listArray否-要读取的表列表可替代table_path例如[{ table_path testdb.table1}, {table_path testdb.table2, query select id, name from testdb.table2}]where_conditionString否-作用于所有表/查询的公共行过滤条件必须以where开头例如where id 100split.sizeInt否8096按表读取时的单个 split 行数表会被拆分为多个 splitsplit.even-distribution.factor.lower-boundDouble否0.05分片列分布因子的下界。分布因子 (MAX(id) - MIN(id) 1) / 行数落入 [下界, 上界] 区间时按均匀分布优化切分低于下界则视为不均匀分布在估计分片数超过split.sample-sharding.threshold时改用采样切分策略split.even-distribution.factor.upper-boundDouble否100分布因子上界语义同上超出上界同样触发采样切分评估split.sample-sharding.thresholdInt否1000触发采样切分策略的估计分片数阈值估算行数 / split.size。超出阈值时启用采样切分以高效处理大表。注意当前仓库源码中该选项默认值为 1000见 JdbcSourceOptions.java 的defaultValue(1000)与文档表格中 10000 的写法不一致建议以源码为准split.inverse-sampling.rateInt否1000采样切分策略的采样率倒数1000 表示 1/1000 采样率数值越大采样越稀疏适合超大表common-options-否-Source 插件公共参数见 Source Common Options 文档docs/en/connectors/common-options/source-common-options.md4.1 分片Split策略的源码印证上述split.*参数并非空配置它们在源码中有明确的实现载体均匀分布判定与采样切分由 DynamicChunkSplitter.java 实现源码注释明确说明maxChunkCount由split.sample-sharding.threshold提供分布因子超出上下界且估计分片数超阈值时切换到基于split.inverse-sampling.rate的采样切分所有默认值split.size 8096、分布因子0.05/100、采样阈值1000、采样率倒数1000均可在 JdbcSourceOptions.java 中以Options.key(...).defaultValue(...)形式逐项核对。4.2 Tips并行与单并发的选择若未设置partition_column任务将以单并发运行设置了partition_column后会按任务并行度并行执行。五、实战配置示例5.1 简单全量读取Simpleenv { parallelism 2 job.mode BATCH } source { Jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://localhost:54321/db_test username root password query select * from source } } transform { # 如需了解更多 transform 插件配置方式可查阅官方 transforms/sql 文档 } sink { Console {} }5.2 并行读取整表Parallel通过配置分片字段实现并行读取适合整表抽取场景。source { Jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://localhost:54321/db_test username root password query select * from source # 并行分片字段 partition_column id # 分片数量 partition_num 10 } }5.3 指定上下边界的并行读取Parallel Boundary显式指定分片列的上下界可以省去 min/max 探测查询读取效率更高。source { Jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://localhost:54321/db_test username root password query select * from source partition_column id partition_num 10 # 读取起始边界 partition_lower_bound 1 # 读取结束边界 partition_upper_bound 500 } }5.4 按 Schema 限定表名Query With Schema NameKingbase 表名通常写作schema.table。连接账号既可以用username也可以使用旧配置键user作为 fallback 被接受。source { Jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://localhost:54321/test user SYSTEM password 123456 query select * from public.e2e_table_source } }5.5 仓库内置 E2E 任务配置参考仓库中 Kingbase 的端到端集成测试提供了可直接参考的完整任务配置Source 读public.e2e_table_sourceSink 批量写入public.e2e_table_sinkjdbc_kingbase_source_and_sink.confenv { parallelism 1 job.mode BATCH } source { jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://e2e_KINGBASEDb:54321/test user SYSTEM password 123456 query select * from public.e2e_table_source } } sink { jdbc { driver com.kingbase8.Driver url jdbc:kingbase8://e2e_KINGBASEDb:54321/test user SYSTEM password 123456 query INSERT INTO public.e2e_table_sink (c1, c2, c3, ...) VALUES (?, ?, ?, ...) } }对应的测试入口为 JdbcKingbaseIT.java它基于 Testcontainers 启动 Kingbase 容器并校验 Source 与 Sink 两侧的列值一致性可作为验证连接器可用性的参照。六、Kingbase 方言的更多实现细节除了读取侧仓库中的 Kingbase 方言实现还覆盖了标识符引用、upsert 与建表等能力理解这些细节有助于排查 DDL/标识符相关问题标识符引用规则KingbaseDialect.java 的quoteIdentifier()对含.的多段名称逐段加双引号如schema.table与 PostgreSQL 风格一致避免大小写敏感与保留字问题tableIdentifier()对库名单独加引号以解决 PG 系数据库名大小写不敏感的历史问题。方言还支持fieldIde字段命名风格配置默认 ORIGINAL。Upsert 语句getUpsertStatement()基于 PostgreSQL 兼容语法生成INSERT ... ON CONFLICT (pk) DO UPDATE SET col EXCLUDED.col ...写侧连接器可直接利用该主键冲突更新能力。建表与表选项校验KingbaseCreateTableSqlBuilder.java 与 KingbaseCatalog.java 实现了 catalog 能力方言仅开放tablespace与fillfactor两个建表选项其中fillfactor会被校验为 10–100 的整数、tablespace会拒绝含引号、换行或分号的值防止 DDL 注入——从源码结构看这些校验失败会抛出CONFIG_VALIDATION_FAILED错误码的配置校验异常。变更历史该连接器自 2.3.4 版本提交 jdbc connector supports Kingbase database (#4803)加入后续随 JDBC 连接器统一演进完整变更记录见 connector-jdbc Changelog。七、小结与适用前提适用前提KingbaseES 8.6 数据源、驱动 jar 已放入$SEATUNNEL_HOME/plugins/jdbc/lib/、JDBC URL 以jdbc:kingbase8:开头引擎可选 Zeta / Flink / Spark运行模式为 BATCH。整表抽取建议优先使用table_path/table_list搭配split.*参数让连接器自动按split.size拆片已有明确主键/自增列且希望控制并发粒度时再使用partition_column系列参数并显式给定partition_lower_bound/partition_upper_bound省去 min/max 探测。类型映射上文档表格之外的TINYINT、MONEY、BLOB、CLOB、BIT以及 MySQL/Oracle 兼容模式类型名已被 KingbaseTypeConverter.java 显式兜底遇到文档未列类型时建议先核对该文件的 switch 分支确认是否已在当前版本支持。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考