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 API（JSON） → 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?loggerLevel=OFF" username = "test" password = "test" generate_sink_sql = true database = "test" table = "public.http_orders" primary_keys = ["c_string"] batch_size = 100 } }

env：并发与运行模式

• parallelism = 1：Http Source 不支持用户自定义分片，且本示例接口为单页返回，单并发即可。
• job.mode = "BATCH"：拉完即退出的批模式。想定时轮询接口就改STREAMING并加poll_interval_millis，e2e 配置 http_streaming_json_to_postgresql.conf 就是这么用的（poll_interval_millis = 5000，每 5 秒轮询一次）。

source：Http Source 为什么必须配 schema？

因为format = "json"时，SeaTunnel 完全依赖schema.fields知道「JSON 里哪些字段、各是什么类型」，才能把原始报文反序列化成结构化行。字段名写错、类型写错都会直接解析失败。

sink：JDBC Sink 两种写入模式，只能选一种

没有 query 为什么任务起不来？看 JdbcSinkFactory.java 的 OptionRule 就明白了：url、driver、schema_save_mode、data_save_mode始终被要求（后两者有默认值可省略）；而generate_sink_sql = true时强制要求database，generate_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;

最易踩的三个问题

进阶玩法

数据不止一页？配置分页拉取

分页能力集中在pageing配置块（注意源码里就是这个拼写），支持两种模式：

PageNumber（默认）：按页码推进。

• page_field：请求里的分页字段名，默认page，可在headers/params/body中用${page}占位；
• start_page_number：起始页码，默认1；
• total_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_field（JSONPath）指定从响应里取游标的位置。当响应游标为空或与当前游标相同时，判定没有更多数据，结束拉取。

💡 排查请求形态的核心思路：先想清楚「最终发出的 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 加rewriteBatchedStatements=true；
• PostgreSQL 大批量导入：use_copy_statement = true走COPY <table> FROM STDIN路径（要求驱动提供getCopyAPI()，且不支持MAP/ARRAY/ROW类型）；
• 精确一次：is_exactly_once = true+xa_data_source_class_name+max_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.conf
• mock 接口定义：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