SeaTunnel 完整指南：HTTP API 数据同步到 PostgreSQL 的端到端实战

SeaTunnel 完整指南HTTP API 数据同步到 PostgreSQL 的端到端实战【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文用 Apache SeaTunnel 完成一条 HTTP 数据同步链路从 HTTP API 拉取 JSON 数据经 Http Source 解析后由 JDBC Sink 写入 PostgreSQL覆盖插件安装、配置、运行到进阶分页与 Upsert 的全过程。你的接口数据为什么想进数据库假设你手上有一个订单查询接口运营每天都要人工导出结果再贴进报表又或者埋点平台按天把统计数据吐出来下游分析平台却只能读数据库。这类「API 出数据、数据库进数据」的需求很常见手工搬运既慢又容易出错。用 SeaTunnel 做这条链路有三个理由Http Source 和 JDBC Sink 都是官方内置连接器整条链路一份 HOCON 配置就能声明完成批模式、流式轮询只是改一个参数的事JDBC Sink 还能自动生成 SQL、自动建表省去手写 DDL。架构总览整条链路长什么样整条管道只有四个节点HTTP APIJSON → Http Source → SeaTunnelRow → JDBC Sink → PostgreSQL 表Http Source发GET/POST请求拉取响应按你声明的schema把 JSON 反序列化成结构化的 SeaTunnelRow可以理解为引擎内部统一的行格式JDBC Sink借助数据库厂商的 JDBC 驱动把行批量写入目标表支持自动生成 INSERT / UPSERT 等 SQL。环境准备清单动手写配置前把下面五件事逐项勾掉。每项都附了自证方式卡住哪一项就先解决哪一项。1. 引擎能跑通先完成跑第一个任务确认安装、配置解析、执行引擎都正常。✅ 自证模板任务config/v2.batch.config.template能完整跑完。2. 按需安装插件2.2.0-beta 起发行包不再默认带全量连接器需要把config/plugin_config收敛成本链路所需的两个插件再执行安装脚本--seatunnel-connectors-- connector-http-base connector-jdbc --end--cd ${SEATUNNEL_HOME} sh bin/install-plugin.sh ls connectors | rg connector-(http-base|jdbc)✅ 自证最后一条命令能列出connector-http-base和connector-jdbc两个目录。插件名与 JAR 的对应关系见仓库根目录的 plugin-mapping.properties。3. 放置 JDBC 驱动SeaTunnel 不统一内置所有驱动各厂商许可证与再分发条款不同Zeta 引擎本地模式需把驱动放进${SEATUNNEL_HOME}/lib并重启相关进程。✅ 自证ls ${SEATUNNEL_HOME}/lib | rg postgresql⚠️ Spark / Flink 引擎则放到每个执行节点的${SEATUNNEL_HOME}/plugins/Jdbc/lib/。常见文件名postgresql-42.x.x.jar、mysql-connector-j-8.x.x.jar、ojdbc8.jar。4. 先看清接口返回了什么curl http://mockserver:1080/example/http本文使用 Http Source 文档配套的 mock 接口mock 定义见 mockserver-config.json。✅ 自证输出 JSON 顶层能看到c_string、c_int等字段且和你打算在 schema 里声明的字段对得上。5. 目标库授权建表配置里开了generate_sink_sql true自动建表所以要给 sink 用户建表权限CREATE USER test WITH PASSWORD test; CREATE DATABASE test OWNER test;重新连接到test库后执行GRANT USAGE, CREATE ON SCHEMA public TO test;✅ 自证用test账号连上test库无报错GRANT执行无权限错误。完整配置与关键参数先把整份配置存为config/http-to-jdbc.conf这是可以直接运行的完整版本env { parallelism 1 job.mode BATCH } source { Http { plugin_output http_orders url http://mockserver:1080/example/http method GET format json schema { fields { c_string string c_int int } } } } sink { Jdbc { plugin_input http_orders driver org.postgresql.Driver url jdbc:postgresql://postgresql:5432/test?loggerLevelOFF username test password test generate_sink_sql true database test table public.http_orders primary_keys [c_string] batch_size 100 } }env并发与运行模式parallelism 1Http Source 不支持用户自定义分片且本示例接口为单页返回单并发即可。job.mode BATCH拉完即退出的批模式。想定时轮询接口就改STREAMING并加poll_interval_millise2e 配置 http_streaming_json_to_postgresql.conf 就是这么用的poll_interval_millis 5000每 5 秒轮询一次。sourceHttp Source 为什么必须配 schema因为format json时SeaTunnel 完全依赖schema.fields知道「JSON 里哪些字段、各是什么类型」才能把原始报文反序列化成结构化行。字段名写错、类型写错都会直接解析失败。参数本篇取值这样设置的原因plugin_outputhttp_orders给本节点输出流命名sink 靠plugin_input指向它完成上下游对接urlhttp://mockserver:1080/example/httpmock 接口地址换成你的真实 API 即可methodGET仅支持GET、POST两种方法formatjson支持json/text/binary默认text设为json时必须配套schemaschema.fieldsc_string string、c_int int只取 mock 返回里的两个顶层字段做最小验证sinkJDBC Sink 两种写入模式只能选一种参数本篇取值这样设置的原因plugin_inputhttp_orders接收 source 节点输出的数据流driverorg.postgresql.Driver与${SEATUNNEL_HOME}/lib中放置的驱动 JAR 对应urljdbc:postgresql://postgresql:5432/test?loggerLevelOFF指向目标 PostgreSQL 实例与test库username/passwordtest/test上面第 5 步创建的账号generate_sink_sqltrue打开自动生成 SQLSeaTunnel 依据 schema 与 RowKind每行是插入、更新还是删除的操作标记生成 INSERT / 原生 UPSERT / UPDATE / DELETEdatabasetestgenerate_sink_sql true时的必填项见下文约束tablepublic.http_orders目标表有 schema 概念的库必须写成schema.table形式primary_keys[c_string]用于生成原生 UPSERT也参与自动建表时的主键约束batch_size100每攒满 100 行触发一次 flush 写入没有 query 为什么任务起不来看 JdbcSinkFactory.java 的 OptionRule 就明白了url、driver、schema_save_mode、data_save_mode始终被要求后两者有默认值可省略而generate_sink_sql true时强制要求databasegenerate_sink_sql false时强制要求query自定义 SQL。generate_sink_sql定义在 JdbcSinkOptions.java 中默认值是false——所以不显式打开它就必须自己写query两种写入模式不能混用。启动任务并核对数据本地模式运行不拉起独立引擎集群适合单机验证cd ${SEATUNNEL_HOME} ./bin/seatunnel.sh --config ./config/http-to-jdbc.conf -m local任务结束后按下面清单核对✅ 日志中无 JSON 解析异常、无 JDBC DDL 报错自动建表语句执行成功✅ 目标表行数与接口返回的记录数一致✅ 字段值与curl直接看到的报文一致SELECT COUNT(*) FROM public.http_orders; SELECT c_string, c_int FROM public.http_orders ORDER BY c_string;最易踩的三个问题现象原因解法任务报 JSON 解析错误或目标表字段大量为 nullschema.fields里字段名拼错、类型与接口返回不匹配format json下 schema 就是反序列化依据用curl逐字段核对返回报文类型体系可参考 Http Source 文档中覆盖map/array/row/decimal/timestamp的完整 schema 示例解析不报错但取不到值行内容为空有效数据被包在深层字段里如{code:200,data:{...}}schema 却直接声明在顶层配置content_field或json_field先做字段抽取详见下文进阶部分同步完只有第一页数据源接口有分页或限流作业却按单页接口处理配置pageing分页块用PageNumber/Cursor推进请求详见下文进阶玩法数据不止一页配置分页拉取分页能力集中在pageing配置块注意源码里就是这个拼写支持两种模式PageNumber默认按页码推进。page_field请求里的分页字段名默认page可在headers/params/body中用${page}占位start_page_number起始页码默认1total_page_size总页数0默认表示未知——此时单页返回行数小于pageing.batch_size默认 100即判定拉完对应 HttpSourceReader.java 中readSize pageInfo.getBatchSize()的终止判断use_placeholder_replacement默认false只做按 key 的整值替换设为true才支持10${page} → 105这类带前后缀的占位替换。Cursor游标page_type Cursor用cursor_field指定请求中携带游标的字段名用cursor_response_fieldJSONPath指定从响应里取游标的位置。当响应游标为空或与当前游标相同时判定没有更多数据结束拉取。 排查请求形态的核心思路先想清楚「最终发出的 HTTP 请求长什么样」。GET下params一定拼进 URL 查询串POST且keep_params_as_form false时 body 按 JSON 发送未配 body 则发空对象{}keep_params_as_form true时 params 并入表单 body 并自动补application/x-www-form-urlencoded头keep_page_param_as_http_param true时分页字段直接写进params。有效数据不在顶层json_field 与 content_field 两种抽取方式两者任选其一即可content_field抽出整个数组或对象片段再交给 schema。例如返回{store:{book:[...]}}配content_field $.store.book.*schema 就只需声明category、author、title等字段。json_field为每个 schema 字段单独指定 JSONPath如json_field { category $.store.book[*].category }必须与schema一起使用。抽取发生在 HttpSourceReader.java 的collect()中先走content_field分支或json_field分支取出目标片段再交给反序列化器。另外 JSON 字段缺失默认抛错设json_filed_missed_return_null true定义于 HttpSourceOptions.java可让缺失字段返回null。重复执行任务数据如何不丢不重靠primary_keys Upsert 两个 SaveMode 参数均有默认值配合Upsert 触发条件SeaTunnel 只有拿到主键/唯一键信息才走 upsert/update 路径。key 依次来自显式primary_keys、上游 Catalog 元数据继承的主键、第一组 unique key都没有就退化为普通 INSERT。Upsert 语句有 key 且enable_upsert true默认时优先用方言原生语句PostgreSQL 生成INSERT ... ON CONFLICT (...) DO UPDATE若所有字段都是主键则退化为DO NOTHING。本篇把随机字符串c_string当主键正因为它能唯一标识记录若确定没有重复 key可设enable_upsert false提速。schema_save_mode默认CREATE_SCHEMA_WHEN_NOT_EXIST表不存在则建、存在则跳过可选RECREATE_SCHEMA删了重建、ERROR_WHEN_SCHEMA_NOT_EXIST不存在报错、IGNORE跳过建表逻辑。data_save_mode默认APPEND_DATA保留已有数据、追加可选DROP_DATA先清空、CUSTOM_PROCESSING配合custom_sql自定义预处理、ERROR_WHEN_DATA_EXISTS有数据即报错。⚠️ 主键选错是最隐蔽的坑它既决定 Upsert 的冲突键也参与自动建表的主键约束选错会导致冲突报错或静默覆盖。吞吐不够怎么办batch_size调大flush 次数减少、吞吐上升但内存占用与故障重试量同步增加MySQL 非 XA 批量任务JDBC URL 加rewriteBatchedStatementstruePostgreSQL 大批量导入use_copy_statement true走COPY table FROM STDIN路径要求驱动提供getCopyAPI()且不支持MAP/ARRAY/ROW类型精确一次is_exactly_once truexa_data_source_class_namemax_retries 0前提是数据库与驱动都支持 XA 事务PostgreSQL 需启用 prepared transaction配置前先确认前置条件。收尾把这条链路套进你的业务这条链路本身是个通用模板改动都很小接口换地址、加鉴权头headers里带上 Token即可对接真实 API目标库换 MySQL / Oracle / SQL Server只需改driver、url和${SEATUNNEL_HOME}/lib下的驱动 JAR需要定时增量就切到STREAMING加poll_interval_millis。理解了参数语义后你完全可以按自己业务的 schema 直接改写这份配置。相关资料Http Source 连接器文档全部源选项、format 详解、分页示例JDBC Sink 连接器文档写入模式、SaveMode、exactly-once、故障排查与驱动参考部署与下载连接器插件跑第一个任务官方 HTTP 到 JDBC 食谱e2e 参考配置http_streaming_json_to_postgresql.confmock 接口定义mockserver-config.json插件映射plugin-mapping.properties关键源码HttpSourceReader.java、HttpSourceOptions.java、JdbcSinkFactory.java、JdbcSinkOptions.java【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考