数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载本文围绕 Apache SeaTunnel 的 OceanBase JDBC Source 连接器展开介绍其引擎支持范围、驱动部署方式、MySQL/Oracle 双兼容模式下的数据类型映射、完整的 Source 参数表以及基于partition_column的分片并行读取机制。通过结合connector-jdbc模块的源码实现读者可以掌握如何在 SeaTunnel含 SeaTunnel Zeta 引擎中配置 OceanBase 数据读取任务并理解底层 JDBC 方言加载与分片切分的完整链路从而正确设计高吞吐的批量同步作业。连接器概览引擎支持与核心能力OceanBase JDBC Source 连接器通过标准的 JDBC 接口读取 OceanBase 外部数据源是一个典型的批式读取batch连接器。根据官方文档 OceanBase.md 的说明它支持以下计算引擎引擎支持情况Spark支持Flink支持SeaTunnel Zeta支持在核心能力维度上该连接器的特性清单如下特性定义参见 connector-v2-features.md✅batch批式读取通过 JDBC 执行查询语句读取全量或指定范围的数据❌stream流式读取暂不支持OceanBase Source 以有界批式数据源接入✅exactly-once精确一次配合 SeaTunnel/Zeta 的 checkpoint 机制保证读取语义✅column projection列投影支持按需选取列减少不必要的字段传输✅parallelism并行度支持多并发并行读取✅support user-defined split用户自定义分片支持通过partition_column等参数自定义分片键与分片数量。从源码结构看该连接器并未单独成模块而是作为 connector-jdbc 通用 JDBC 连接器的一部分实现OceanBase 的方言、目录Catalog与驱动依赖均挂在 JDBC 连接器之下因此任何使用 JDBC 源插件配置插件名为Jdbc的作业只要指定 OceanBase 驱动与 URL 即可接入。工作原理双兼容模式下的方言分发OceanBase 数据库同时支持 MySQL 兼容模式与 Oracle 兼容模式这决定了它的 SQL 方言、元数据查询方式与数据类型体系都需要按模式区分。SeaTunnel 在源码层面对这一差异做了明确处理。JdbcSourceFactory在创建 Source 时会调用JdbcDialectLoader.load(url, compatibleMode)加载方言见 JdbcSourceFactory.java。加载过程见 JdbcDialectLoader.java使用 Java SPIServiceLoader机制发现所有JdbcDialectFactory实现再根据 URL 前缀匹配出唯一能处理该 URL 的工厂。针对 OceanBase 的方言工厂 OceanBaseDialectFactory.java 实现了acceptsURL只要 URL 以jdbc:oceanbase:开头即命中该工厂而它的create(compatibleMode, fieldIde)方法会依据兼容模式分发compatible_mode oracle→ 复用OracleDialectOracle 方言其他值即mysql→ 复用MysqlDialectMySQL 方言。值得特别注意的是该工厂的无参create()直接抛出UnsupportedOperationException提示 “Cant create JdbcDialect without compatible mode for OceanBase”。这说明对 OceanBase 而言compatible_mode虽然在整个 JDBC 源插件的参数规则JdbcSourceFactory.java 中将COMPATIBLE_MODE声明为可选中属于可选但在 OceanBase 场景下必须显式指定否则无法创建方言。这也解释了官方文档将其标记为 “Required” 的原因。与 Source 侧对应OceanBase 的 Catalog 工厂 OceanBaseCatalogFactory.java 同样校验compatibleMode配置缺失时直接抛出 “Miss config !”并据此创建OceanBaseMySqlCatalog或OceanBaseOracleCatalog用于表结构schema的获取与元数据管理。环境准备驱动依赖与插件部署要运行 OceanBase Source 任务首先需要准备 OceanBase 的 JDBC 驱动。官方文档在 OceanBase.md 的 Database Dependency 一节给出了明确步骤从 Maven 仓库下载com.oceanbase:oceanbase-client对应版本的驱动 jar将 jar 复制到 SeaTunnel 安装目录的插件库路径下即$SEATUNNEL_HOME/plugins/jdbc/lib/示例命令cp oceanbase-client-xxx.jar $SEATUNNEL_HOME/plugins/jdbc/lib/从仓库的 connector-jdbc/pom.xml 可以看到oceanbase-client以provided作用域引入当前仓库锁定的版本为2.4.3对应oceanbase.jdbc.version即驱动不随连接器打包、需运行时自行提供这正是文档要求手动复制 jar 的源码依据。部署时请根据实际 OceanBase 集群版本选择兼容的驱动版本。数据源信息与连接参数官方文档给出的 OceanBase 数据源支持信息如下数据源支持版本驱动类JDBC URLMaven 坐标OceanBase所有 OceanBase 服务端版本com.oceanbase.jdbc.Driverjdbc:oceanbase://localhost:2883/testcom.oceanbase:oceanbase-client其中端口2883是 OceanBase 的默认 RPC/OBProxy 端口URL 中的test为要读取的数据库名。连接参数compatible_mode取值mysql或oracle必须与目标租户实际的兼容模式一致否则方言与元数据解析都会出错。数据类型映射OceanBase 数据通过 JDBC 驱动读出后SeaTunnel 会将其转换为自身的 SeaTunnel Row 类型体系。由于双兼容模式的存在映射规则也分为两套以下两张表为官方文档的完整映射内容。MySQL 兼容模式OceanBaseMySQL 模式数据类型SeaTunnel 数据类型BIT(1)、TINYINT(1)BOOLEANTINYINTBYTETINYINT、TINYINT UNSIGNEDSMALLINTSMALLINT UNSIGNED、MEDIUMINT、MEDIUMINT UNSIGNED、INT、INTEGER、YEARINTINT UNSIGNED、INTEGER UNSIGNED、BIGINTBIGINTBIGINT UNSIGNEDDECIMAL(20,0)DECIMAL(x,y)列精度 38DECIMAL(x,y)原样保留精度DECIMAL(x,y)列精度 38DECIMAL(38,18)DECIMAL UNSIGNEDDECIMAL(精度1, 小数位)FLOAT、FLOAT UNSIGNEDFLOATDOUBLE、DOUBLE UNSIGNEDDOUBLECHAR、VARCHAR、TINYTEXT、MEDIUMTEXT、TEXT、LONGTEXT、JSON、ENUMSTRINGDATEDATETIMETIMEDATETIME、TIMESTAMPTIMESTAMPTINYBLOB、MEDIUMBLOB、BLOB、LONGBLOB、BINARY、VARBINARY、BIT(n)、GEOMETRYBYTES映射要点无符号整数自动升型MySQL 模式下TINYINT UNSIGNED无法容纳于BYTE因此映射到SMALLINTINT UNSIGNED、BIGINT映射到BIGINT而BIGINT UNSIGNED超出有符号 64 位范围进一步映射到DECIMAL(20,0)。高精度十进制截断DECIMAL(x,y)精度超过 38 时按DECIMAL(38,18)处理进行截断式兼容。字符串家族广泛收敛为STRING包括文本类型与JSON、ENUM。二进制与几何类型收敛为BYTES包括各类 BLOB、二进制字符串与GEOMETRY。Oracle 兼容模式OceanBaseOracle 模式数据类型SeaTunnel 数据类型IntegerDECIMAL(38,0)Number(p)p ≤ 9INTNumber(p)p ≤ 18BIGINTNumber(p)p 18DECIMAL(38,18)Number(p,s)DECIMAL(p,s)FloatDECIMAL(38,18)REAL、BINARY_FLOATFLOATBINARY_DOUBLEDOUBLECHAR、NCHAR、VARCHAR、VARCHAR2、NVARCHAR2、NCLOB、CLOB、LONG、XML、ROWIDSTRINGDATETIMESTAMPTIMESTAMP、TIMESTAMP WITH LOCAL TIME ZONETIMESTAMPBLOB、RAW、LONG RAW、BFILEBYTESUNKNOWN暂不支持映射要点Number 按精度分档Oracle 的Number类型按精度 p 动态映射为INTp≤9、BIGINTp≤18、DECIMAL(38,18)p18带小数位的Number(p,s)则原样映射为DECIMAL(p,s)Oracle 的Integer与Float也被纳入十进制体系DECIMAL(38,0)与DECIMAL(38,18)这与 MySQL 模式下FLOAT/DOUBLE直接映射为浮点类型形成对比体现了 Oracle 模式对数值精度的保守处理。DATE 映射为 TIMESTAMPOracle 的DATE包含时间部分因此映射为TIMESTAMP而非DATE。UNKNOWN暂不支持无法识别的类型会映射失败需在 SQL 查询中显式转换。Source 参数详解OceanBase Source 的全部可配置参数如下表与官方文档保持一致并补充了源码层面的取值约束参数名类型必填默认值说明urlString是-JDBC 连接地址例如jdbc:oceanbase://localhost:2883/testdriverString是-驱动类名OceanBase 场景固定为com.oceanbase.jdbc.DriveruserString否-连接实例的用户名passwordString否-连接实例的密码compatible_modeString是OceanBase 场景-OceanBase 兼容模式取值为mysql或oracle。如前文源码分析所述缺失时方言工厂将无法创建方言queryString是-数据读取的查询语句connection_check_timeout_secInt否30连接有效性校验等待时长秒。对应源码 JdbcOptions.java 中connection_check_timeout_sec的默认值 30partition_columnString否-并行分片所依据的列名仅支持数值类型列与字符串类型列partition_lower_boundBigDecimal否-分片扫描的下边界不设置时 SeaTunnel 会查询数据库自动获取最小值partition_upper_boundBigDecimal否-分片扫描的上边界不设置时 SeaTunnel 会查询数据库自动获取最大值partition_numInt否作业并行度分片数量仅支持正整数默认取作业的并行度fetch_sizeInt否0每次查询抓取的行数用于减少数据库往返次数、提升大结果集读取性能0 表示使用 JDBC 驱动默认值。对应源码 JdbcOptions.java 中fetch_size的定义propertiesMap否-额外的连接配置参数。当properties与 URL 中携带相同参数时优先级由驱动具体实现决定例如 MySQL 驱动中properties优先于 URLcommon-options-否-Source 插件通用参数详见 Source Common Options主要包括result_table_name注册临时表供下游通过source_table_name引用与parallelism覆盖 env 中的并行度并行度行为提示官方文档 Tips 明确指出如果未设置partition_column任务将退化为单并发执行设置了partition_column后任务将根据并发度进行并行分片读取。也就是说partition_column是并行读取的“开关”只有指定了分片列JdbcSourceSplitEnumerator才会基于该列生成多个分片split分发给各并行 reader。并行读取与分片原理源码级在理解上述参数后可以进一步深入connector-jdbc的分片实现来印证其行为分片驱动链JdbcSourceFactory将partition_column、partition_num、partition_lower_bound、partition_upper_bound等参数解析进JdbcSourceConfig参见 JdbcSourceConfig.java随后由JdbcSourceSplitEnumerator负责把整个查询区间切分为JdbcSourceSplit列表。分片消费每个并行实例对应一个JdbcSourceReader见 JdbcSourceReader.java它内部维护一个分片队列splits在pollNext中逐个取出分片交给JdbcInputFormat执行查询并产出SeaTunnelRow当队列耗尽且无更多分片时调用signalNoMoreElement结束有界数据源。snapshotState会快照剩余分片配合 checkpoint 保证exactly-once语义。边界自动探测partition_lower_bound/partition_upper_bound未设置时连接器会通过SELECT MIN(partition_column)、SELECT MAX(partition_column)自动获取列的最小/最大值作为扫描区间因此文档中“Parallel 示例”无需手动指定边界即可整表分片读取。进阶分片参数源码 JdbcSourceOptions.java 中还定义了若干更精细的分片策略参数属于 JDBC 源通用能力官方 OceanBase 文档未逐项列出可按需选用split.size默认 8096单分片的行数规模分片切分以行数估算为基础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在数据分布因子超出上下界且估算分片数超过阈值时触发采样分片策略采样率 1/1000用于高效处理超大表。连接池与校验连接管理基于 HikariCP在 connector-jdbc/pom.xml 中以com.zaxxer:HikariCP引入并做了 shade 重定位以避免与 Spark 环境冲突connection_check_timeout_sec即用于连接有效性校验的超时控制。任务配置示例以下三个示例完整取自官方文档可直接套用。1. 简单读取单并发env { parallelism 2 job.mode BATCH } source { Jdbc { driver com.oceanbase.jdbc.Driver url jdbc:oceanbase://localhost:2883/test?useUnicodetruecharacterEncodingUTF-8rewriteBatchedStatementstrue user root password compatible_mode mysql query select * from source } } transform { # If you would like to get more information about how to configure seatunnel and see full list of transform plugins, # please go to https://seatunnel.apache.org/docs/transform/sql } sink { Console {} }说明该示例未设置partition_column因此虽然env.parallelism 2实际仍以单并发读取整表URL 中的rewriteBatchedStatementstrue与characterEncodingUTF-8属于推荐的连接优化参数。2. 并行分片读取ParallelRead your query table in parallel with the shard field you configured and the shard data. You can do this if you want to read the whole tableenv { parallelism 10 job.mode BATCH } source { Jdbc { driver com.oceanbase.jdbc.Driver url jdbc:oceanbase://localhost:2883/test?useUnicodetruecharacterEncodingUTF-8rewriteBatchedStatementstrue user root password compatible_mode mysql query select * from source # 并行分片读取字段 partition_column id # 分片数量 partition_num 10 } } sink { Console {} }说明指定partition_column id后连接器按id列将全表数据切分为 10 个分片并行读取partition_num应与env.parallelism相匹配以获得最佳吞吐。上、下边界未指定时由连接器自动探测MIN(id)/MAX(id)。3. 自定义分片边界Parallel BoundaryIt is more efficient to read your data source according to the upper and lower boundaries you configuredsource { Jdbc { driver com.oceanbase.jdbc.Driver url jdbc:oceanbase://localhost:2883/test?useUnicodetruecharacterEncodingUTF-8rewriteBatchedStatementstrue user root password compatible_mode mysql query select * from source partition_column id partition_num 10 # 读取起始边界 partition_lower_bound 1 # 读取结束边界 partition_upper_bound 500 } }说明显式指定上下边界可避免连接器额外执行MIN/MAX探测查询对已知数据范围的场景更高效边界值为BigDecimal适用于数值型分片列。4. 面向生产环境的推荐配置在官方示例基础上可结合前文参数组合出一份更完整的生产配置分片键使用数字主键、显式边界、加大抓取批量并补充连接属性env { parallelism 8 job.mode BATCH } source { Jdbc { driver com.oceanbase.jdbc.Driver url jdbc:oceanbase://localhost:2883/test user root password your_password compatible_mode oracle # 按目标租户实际模式填写 mysql / oracle query select id, name, created_at from source partition_column id partition_num 8 partition_lower_bound 1 partition_upper_bound 1000000 fetch_size 1000 # 每批抓取行数0 为驱动默认 connection_check_timeout_sec 30 properties { useUnicode true characterEncoding UTF-8 rewriteBatchedStatements true } result_table_name oceanbase_source # Source 通用参数注册临时表 } } transform { # 可选Sql 等 transform 插件 } sink { Console { source_table_name oceanbase_source } }实践建议与注意事项务必正确设置compatible_mode这是 OceanBase 连接器区别于其他 JDBC 源的强制项mysql与oracle两套方言、元数据与类型映射完全不同填错会导致方言加载失败或数据解析异常。这也是 OceanBaseDialectFactory.java 在缺少兼容模式时直接抛异常的根因。并行读取的前提是分片列不设partition_column即单并发设置后才会按partition_num默认作业并行度切分。分片列建议选择分布均匀的数值主键配合显式的partition_lower_bound/partition_upper_bound可省去边界探测查询。驱动 jar 必须手动部署oceanbase-client在连接器中为provided作用域仓库版本为 2.4.3需下载对应版本并复制到$SEATUNNEL_HOME/plugins/jdbc/lib/否则运行时会报找不到驱动类。类型映射需在写库前确认高精度DECIMAL精度 38会被截断为DECIMAL(38,18)BIGINT UNSIGNED映射为DECIMAL(20,0)Oracle 模式的DATE映射为TIMESTAMP——在对接下游 Sink 时应按目标端能力合理选择字段类型避免精度丢失。大表性能调优方向可从fetch_size减少网络往返、partition_num提升并行度、以及split.size等分片策略参数入手数据分布极度不均时可关注采样分片策略split.sample-sharding.threshold、split.inverse-sampling.rate的触发逻辑。有界批式语义该连接器不支持流式CDC读取需要增量同步时应改用 MySQL-CDC 等 CDC 类连接器方案。通过本文的配置示例与源码佐证读者可以基于 SeaTunnelJdbc源插件快速搭建 OceanBase 到任意目标端的批量数据同步管道并在遇到方言、类型或性能问题时直接定位到 connector-jdbc 模块的方言、选项与分片实现进行排查。赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐MiniMax-Music3 音乐生成全解析DiffSynth-Studio 中的两阶段级联推理与低显存部署MiniMax Music3 音乐生成全解析DiffSynth Studio 中的两阶段级联推理与低显存部署 MiniMax Music3 是 DiffSyn数据工程大数据批处理流处理OneUptime 值班通知电话号码白名单配置指南SMS 与电话告警送达保障OneUptime 值班通知电话号码白名单配置指南SMS 与电话告警送达保障 导读 OneUptime 通过 Twilio 等渠道向值班工程师On Call数据工程大数据批处理流处理SeaTunnel Greenplum Source 连接器完整指南基于 JDBC 的数据读取、驱动配置与并行分片原理SeaTunnel Greenplum Source 连接器完整指南基于 JDBC 的数据读取、驱动配置与并行分片原理 SeaTunnel 通过 Jdbc S云原生上一篇MXNet 在树莓派上的构建与安装指南交叉编译、pip 安装与 ARM 原生构建全流程下一篇如何快速掌握ECharts桑基图布局优化5个技巧彻底告别节点重叠创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考