SeaTunnel Phoenix Sink 实战指南基于 Jdbc 连接器将数据 UPSERT 写入 HBase【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本指南以 SeaTunnel 仓库中 Phoenix 数据接收器文档 为主线深入讲解如何通过Jdbc连接器把数据写入 Apache Phoenix 表包括 thick / thin 两种 JDBC 驱动的选型与 URL 配置、upsert语句的参数绑定规则、完整可运行的 HOCON 任务示例并结合connector-jdbc模块的 Phoenix 方言源码与 E2E 测试配置说明其底层实现原理。读完本文你将能独立完成「任意上游数据源 → Phoenix(HBase)」的 SeaTunnel 同步任务配置与排障。概述Phoenix Sink 是如何工作的Phoenix Sink 并不是一个独立的连接器实现而是复用 Jdbc 连接器在作业配置中以Jdbc作为连接器标识符。其核心写入链路是通过 Phoenix 的 JDBC 驱动执行upsert语句将每一行数据写入 HBase。Phoenix 提供 SQL 层接口把对 HBase 的读写封装成标准的 JDBC 操作因此 SeaTunnel 只需依赖connector-jdbc模块的 JDBC 写入能力参数化语句、批量提交等配合 Phoenix 方言即可完成对接。当前仓库已测试的 Phoenix 版本为 4.x 和 5.x。支持的引擎Phoenix Sink 支持以下引擎运行SparkFlinkSeaTunnel Zeta两种 JDBC 驱动thick 与 thin使用 Java JDBC 连接 Phoenix 有两条路径二者的驱动类名和连接 URL 均不同这是配置 Phoenix Sink 时首先要确定的事情驱动类型连接方式driver 配置url 配置示例thick 驱动直接连接 ZooKeeper 集群org.apache.phoenix.jdbc.PhoenixDriverjdbc:phoenix:localhost:2182/hbasethin 驱动连接 Phoenix Query Serverorg.apache.phoenix.queryserver.client.Driverjdbc:phoenix:thin:urlhttp://localhost:8765;serializationPROTOBUF提示 1SeaTunnel 默认使用 thin 驱动 jar。如果需要使用 thick 驱动或者其他版本的 Phoenix thin 驱动需要重新编译connector-jdbc模块将对应的驱动依赖打入该模块。提示 2当前接收器不支持精确一次exactly-once语义因为 Phoenix 暂不支持 XA 事务。参见 连接器特性说明 中的精确一次定义。从源码层面看连接器正是通过 URL 前缀来识别 Phoenix 方言的。在 PhoenixDialectFactory.java 中Override public boolean acceptsURL(NonNull String url) { return url.startsWith(jdbc:phoenix:); }也就是说只要url以jdbc:phoenix:开头thick 与 thin 两种 URL 都满足JDBC Sink 就会自动选用PhoenixDialect进行类型映射与行转换无需额外配置dialect参数。主要特性特性支持情况精确一次❌ 不支持Phoenix 不支持 XA 事务需要特别说明的是通用 Jdbc Sink 文档中宣称的 exactly-once基于 XA 事务能力在 Phoenix 场景下不可用Phoenix 及其 JDBC 驱动不提供 XA 数据源支持。因此 Phoenix Sink 实际提供的是至多一次 / 至少一次的批量写入能力在任务重跑时可能出现重复 UPSERT但由于upsert本身以主键为幂等键重复执行通常不会产生重复数据取决于具体业务主键设计。选项Options名称类型是否必填默认值描述driverString是-JDBC 驱动类。thick 驱动使用org.apache.phoenix.jdbc.PhoenixDriverthin 驱动使用org.apache.phoenix.queryserver.client.Driver。urlString是-JDBC 连接 URL。thick 驱动使用jdbc:phoenix:localhost:2182/hbasethin 驱动使用jdbc:phoenix:thin:urlhttp://localhost:8765;serializationPROTOBUF。queryString是-写入数据时执行的 Phoenix upsert 语句例如upsert into test.sink(age, name) values(?, ?)。?占位符会按位置绑定到上游行字段。common-options否-接收器插件通用参数详见 Sink 通用选项。driver [string]JDBC 驱动类名二者必选其一thick 驱动org.apache.phoenix.jdbc.PhoenixDriver直连 ZooKeeper适合能直接访问 HBase 集群 ZK 节点的场景thin 驱动org.apache.phoenix.queryserver.client.Driver经由 Phoenix Query Server默认端口 8765转发请求适合网络隔离更严格、需要统一接入层的集群。url [string]JDBC 连接 URL与 driver 一一对应thickjdbc:phoenix:localhost:2182/hbase格式为jdbc:phoenix:zkQuorum/hbaseRootNode2182是 ZooKeeper 端口/hbase是 HBase 在 ZK 中的根节点thinjdbc:phoenix:thin:urlhttp://localhost:8765;serializationPROTOBUF其中http://localhost:8765指向 Phoenix Query ServerserializationPROTOBUF指定序列化协议也可使用serializationJSON但需服务端配合。query [string]写入数据时执行的 Phoenix upsert 语句例如upsert into test.sink(age, name) values(?, ?)。需要注意?占位符会按位置绑定到上游行字段因此 upsert 中列的顺序需要和上游schema.fields的字段顺序严格一致表名必须使用带 schema 的完全限定名例如test.sink不能只写表名。该参数与 Jdbc 连接器的「用户提供 SQL」写入模式对应generate_sink_sql保持默认false时query为必填。通用 JDBC 写入器AbstractJdbcRowConverter体系会按上游行字段顺序把值依次 set 到各?占位符上Phoenix 场景下由 PhoenixJdbcRowConverter.java 负责行级转换其内部继承自通用的AbstractJdbcRowConverter因此字段类型到 JDBC 类型的绑定行为与其他 Jdbc 方言保持一致。common options接收器插件通用参数例如source_table_name、parallelism等详见 Sink 通用选项。任务示例使用 thick 驱动env { parallelism 1 job.mode BATCH } source { FakeSource { row.num 2 schema { fields { age int name string } } rows [ { kind INSERT, fields [10, jared] } { kind INSERT, fields [20, huan] } ] } } sink { Jdbc { driver org.apache.phoenix.jdbc.PhoenixDriver url jdbc:phoenix:localhost:2182/hbase query upsert into test.sink(age, name) values(?, ?) } }使用 thin 驱动env { parallelism 1 job.mode BATCH } source { FakeSource { row.num 2 schema { fields { age int name string } } rows [ { kind INSERT, fields [10, jared] } { kind INSERT, fields [20, huan] } ] } } sink { Jdbc { driver org.apache.phoenix.queryserver.client.Driver url jdbc:phoenix:thin:urlhttp://spark_e2e_phoenix_sink:8765;serializationPROTOBUF query upsert into test.sink(age, name) values(?, ?) } }两个示例中上游FakeSource的schema.fields顺序为age, name与query中values(?, ?)的列顺序一一对应第一个?绑定ageint第二个?绑定namestring。写入前请确保 Phoenix 侧目标表已存在例如CREATE TABLE test.sink ( age INTEGER PRIMARY KEY, name VARCHAR(255) );由于 Phoenix 以主键为 upsert 的判定依据建议目标表设置与业务一致的PRIMARY KEY以便重复数据能走覆盖更新而非报错。源码实现纵深Phoenix 方言在 JDBC 连接器中的落地为帮助理解底层行为这里补充connector-jdbc模块中 Phoenix 方言实现的几个关键证据1. 方言注册与识别PhoenixDialectFactory.java 通过AutoService(JdbcDialectFactory.class)注册acceptsURL判断 URL 是否以jdbc:phoenix:开头。因此两种驱动 URL 都能被正确路由到 Phoenix 方言。2. 方言不提供自动 upsert 语句PhoenixDialect.java 中getUpsertStatement(...)返回Optional.empty()这意味着 Phoenix 方言不参与自动生成 upsert SQL 的路径——这与本文档要求必须显式提供query参数的事实相互印证Phoenix 写入必须由用户手写 upsert 语句。3. 类型映射PhoenixTypeConverter.java 定义了 SeaTunnel 类型与 Phoenix 原生类型的双向转换规则数值类型TINYINT/UNSIGNED_TINYINT → BYTESMALLINT/UNSIGNED_SMALLINT → SHORTINTEGER/UNSIGNED_INT → INTBIGINT/UNSIGNED_LONG → LONGDECIMAL/FLOAT → FLOATDOUBLE → DOUBLE字符串类型CHAR/VARCHAR → STRING其中 VARCHAR 长度上限为10485760见常量MAX_VARCHAR_LENGTH时间类型DATE → LOCAL_DATETIME → LOCAL_TIMETIMESTAMP → LOCAL_DATE_TIMETIME/TIMESTAMP 的 scale 上限为 6超出会被截断并告警二进制类型BINARY/VARBINARY → BYTES数组类型Phoenix 的ARRAY被映射为ArrayType.STRING_ARRAY_TYPE反向转换时支持 BOOLEAN/TINYINT/SMALLINT/INT/BIGINT/FLOAT/DOUBLE/STRING 等元素类型。该转换器被 PhoenixTypeMapper.java 引用用于把 JDBCResultSetMetaData的列信息转换为 SeaTunnel 的Column描述。这意味着类型映射不仅影响写入也同时服务于 Phoenix 作为 JDBC Source 的读取场景。4. E2E 测试验证仓库在 JdbcPhoenixIT.java 中提供了基于 Testcontainers镜像seatunnelhub/hbase-phoenix-docker:1.0的集成测试启动 Phoenix Query Server容器端口 8765创建test.SOURCE/test.SINK两张表再通过 jdbc_phoenix_source_and_sink.conf 运行「JDBC Source 读取 → JDBC Sink upsert 写入」的完整链路source { Jdbc { driver org.apache.phoenix.queryserver.client.Driver url jdbc:phoenix:thin:urlhttp://seatunnel_e2e_phoenix:8765;serializationPROTOBUF query select * from test.SOURCE } } sink { Jdbc { driver org.apache.phoenix.queryserver.client.Driver url jdbc:phoenix:thin:urlhttp://seatunnel_e2e_phoenix:8765;serializationPROTOBUF query upsert into test.SINK(age, name) values(?, ?) } }注意测试中目标表以age INTEGER PRIMARY KEY, name VARCHAR(255)定义与 upsert 语句的列顺序一致再次印证「query 列顺序必须与上游字段顺序一致」这一约束。该测试可作为搭建本地 Phoenix 写入环境的可复现参考。使用建议与注意事项驱动选择默认 thin 驱动开箱即用若你的网络环境要求直连 ZooKeeper需自行编译connector-jdbc模块引入 thick 驱动并按 Jdbc 连接器文档 中「使用依赖」一节把驱动 JAR 放到对应引擎目录Spark/Flink 放入${SEATUNNEL_HOME}/plugins/Jdbc/lib/Zeta 放入${SEATUNNEL_HOME}/lib/并重启进程。schema 与列顺序query中列的书写顺序必须与上游schema.fields顺序一致否则会发生字段错位写入。完全限定表名upsert 语句中的表名必须携带 schema例如test.sink不能只写sink。不支持 exactly-oncePhoenix 无 XA 支持任务使用至少一次语义利用 Phoenixupsert的主键覆盖特性可缓解重复写入影响。目标表需预先存在Phoenix 场景推荐显式提供query用户提供 SQL 模式该模式下schema_save_mode/data_save_mode等 SaveMode 配置不生效目标表需在任务启动前手动创建。变更日志本连接器复用connector-jdbc的变更日志详见 connector-jdbc 变更日志。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
