企业官网建设流程全解析

SeaTunnel 7 步跑通：把 HTTP API 数据同步进关系型数据库，自动建表和 Upsert 都省了

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

你想把业务系统的 HTTP API 返回数据周期性同步进关系型数据库，不用自己写调度脚本：SeaTunnel 把Http Source和JDBC Sink串起来，一条配置文件就能完成整条数据同步链路。

为什么是这条链路

业务数据经常散落在各个 HTTP API 里：订单服务、营销平台、第三方数据供应商。下游报表和分析的需求高度一致——按周期把数据落到关系型数据库，供 BI 侧查询。传统做法是自己写脚本：请求接口、解析 JSON、处理分页与重试、拼 INSERT SQL、管批次和事务、失败还得重启。

SeaTunnel 把这条链路拆成两个连接器：Http Source 负责拉取响应并按 schema 反序列化成结构化行；JDBC Sink 负责生成目标表 DDL 与写入 SQL，顺带自动建表和主键 Upsert。中间你只写声明式配置，不用写一行 Java，失败重试交给引擎自己处理。

对「接个 API 进数仓」这种小需求，这是最轻的方案：不引入调度系统，不经中间队列，一个配置文件就能跑。

一张图看懂数据流向

先看一行箭头：

HTTP API (JSON) ── Http Source ──> SeaTunnelRow ── JDBC Sink ──> PostgreSQL 表

链路上发生三件事：

• Http Source：发起 GET/POST 请求，把 JSON 响应按schema声明反序列化成 SeaTunnelRow 结构化行；
• 中间模型：引擎内的 SeaTunnelRow 是连接器之间的标准行格式，字段由schema.fields声明，上下游靠它对齐；
• JDBC Sink：按batch_size攒批，按目标库方言生成 INSERT 或原生 UPSERT，写库前还能自动建表。

这条链路在 SeaTunnel 整体架构中的位置：

🔌 从零跑通（动手区）

你接下来要做 6 件事。

1. 确认基础环境能起来

前提是你已经跑过 跑第一个任务，它一次性验证了安装与配置解析链路，这条链路只多换两个连接器，环境要求完全一样。

2. 把插件收敛到只装两个

2.2.0-beta 起二进制发行包不再默认附带全部连接器依赖。先把${SEATUNNEL_HOME}/config/plugin_config改成下面这样，再执行安装脚本并确认 JAR 落盘：

--seatunnel-connectors-- connector-http-base connector-jdbc --end--

cd "${SEATUNNEL_HOME}" sh bin/install-plugin.sh ls connectors | rg 'connector-(http-base|jdbc)'

为什么要收敛：plugin_config里的插件名和 JAR 名一一对应，可对照根目录的 plugin-mapping.properties 核对（seatunnel.source.Http = connector-http-base、seatunnel.sink.Jdbc = connector-jdbc）。

3. 放置 JDBC 驱动

SeaTunnel 不内置所有数据库驱动（各厂商许可证与再分发条款不同）。Zeta 引擎（本地模式）把 PostgreSQL 驱动 JAR 放进${SEATUNNEL_HOME}/lib并重启受影响进程，驱动才会进类路径；Spark / Flink 引擎则放到每个执行节点的${SEATUNNEL_HOME}/plugins/Jdbc/lib/。

ls "${SEATUNNEL_HOME}/lib" | rg 'postgresql'

4. 先看 JSON 再写 schema

别急着写schema，先用 curl 看一眼响应长什么样。这里直接用 Http Source 文档里的示例接口：

curl http://mockserver:1080/example/http

响应顶层能看到c_string、c_int、c_bigint、c_timestamp等 key，示例 schema 只取c_string和c_int。mock 数据定义在 mockserver-config.json（匹配GET /example/http）。如果你的真实接口把数据包在data这类深层字段里，先配好content_field或json_field再运行，否则 schema 解析必挂。

5. 建 PostgreSQL 目标库并授权

配置用了generate_sink_sql = true，SeaTunnel 会自己发 DDL，所以 sink 账号需要publicschema 下建表的权限：

CREATE USER test WITH PASSWORD 'test'; CREATE DATABASE test OWNER test;

重新连接到test库后再执行：

GRANT USAGE, CREATE ON SCHEMA public TO test;

6. 写配置并运行

保存为${SEATUNNEL_HOME}/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 } }

单并发、批模式、两个连接器靠plugin_output/plugin_input对接，这就是最小可运行组合。

cd "${SEATUNNEL_HOME}" ./bin/seatunnel.sh --config ./config/http-to-jdbc.conf -m local

-m local表示本地模式运行，不启动独立引擎集群，适合单机验证。seatunnel.sh入口会根据参数把命令派发到对应执行器，派发路径如下图所示：

配置参数按功能域速查

不按 env / source / sink 三大块拆，改成按「这个参数实际在干什么」分组，方便你对着配置查问题。

数据流命名与连接

plugin_output和plugin_input是数据流两端的显式契约：source 给自己的输出起名，sink 用同名指过来，对接完成。url是请求地址；method只支持GET和POST，默认GET（枚举定义在HttpSourceOptions）；format声明响应体类型，json/text/binary三选一，默认text。

反序列化契约

format = "json"时schema是强制项——HttpSourceFactory.java 的 OptionRule 里有一条条件分支：format 为 JSON 时强制要求schema。schema.fields是反序列化依据，字段名和类型都要跟响应 key 逐一对上。接口把数据包在深层时，补content_field（整段抽取数组或对象片段）或json_field（按字段给 JSONPath，必须搭配schema）。这里有个坑：json_filed_missed_return_null默认false，字段缺失直接抛错；接口字段不稳定、想缺了就填null，要显式设成true。

目标库写入行为

直接说结论：JDBC Sink 有两种互斥的写入模式，由generate_sink_sql（默认false）决定。JdbcSinkFactory.java 的 OptionRule 用两条 conditional 写死了这个约束：为true强制要求database，为false强制要求query，两种模式不可混用。自动 SQL 模式下，database和table定位目标，有 schema 概念的数据库要把table写成xxx.xxx形式（示例是public.http_orders）；primary_keys既参与建表主键约束，也参与 Upsert；batch_size是攒满多少行触发一次 flush（sink 侧默认 1000，示例里设了 100）。

两个 SaveMode 参数都有默认值，通常可省：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（有数据即报错）。

执行环境

parallelism = 1不是保守，是上限：Http Source 是单分片 reader（HttpSourceReader继承AbstractSingleSplitReader），水平拆不开，调高并发也不会拉得更快。真正的选择是job.mode：BATCH拉完即退出；接口是增量返回、要定时轮询时切STREAMING并在 source 上配poll_interval_millis，e2e 配置 http_streaming_json_to_postgresql.conf 就是这个用法（5 秒一轮）。

🔍 核对结果 + 排错决策树

任务结束后跑两条 SQL：

SELECT COUNT(*) FROM public.http_orders; SELECT c_string, c_int FROM public.http_orders ORDER BY c_string;

行数应该和接口返回体的对象数量一致，取值与 curl 输出逐条对得上。对不上就按下面的树往下走：

• 任务报错或行数不符
  • 日志报解析 / 反序列化错误
    • 先查schema.fields：字段名和真实响应 key 是否一致、类型是否对得上（curl 一次核对）
    • 名字都对但取不到值：接口返回的是嵌套结构，没配content_field/json_field，schema 在顶层空转
  • 任务成功但行数明显偏少
    • 源接口有分页，作业只取了第一页？（没配pageing块）
    • 配了total_page_size的，数值是否小于实际总页数
  • 自动建表成功但写入反复失败、或数据被莫名覆盖
    • primary_keys选的列能否唯一标识一行？键选错，upsert 时要么主键冲突要么静默覆盖，数据从此不符合预期
  • HTTP 状态码非 2xx
    • url、method或鉴权 header 有问题，拿日志里的状态码和响应体与 curl 结果比对

三个进阶场景（扁平展开）

嵌套 JSON 怎么抽

响应长这样：{"code":200,"data":[{...},{...}]}，schema 只看顶层，永远碰不到data里的字段。两种抽法按场景选：

• content_field：直接指向数组或对象片段，如content_field = "$.data.*"，连接器只把data数组交给 schema 解析，schema.fields只需声明id、name这类字段。适合「有效负载整体是个数组、字段和 schema 一一对应」的场景。
• json_field：按字段逐个给 JSONPath，如json_field = { id = "$.data[*].id", name = "$.data[*].name" }，必须和schema搭配。适合目标字段散落在不同分支、只挑几列拿的场景。

两条路径对应 HttpSourceReader.javacollect()里的getPartOfJson（content_field 分支）与decodeJSON+parseToMap（json_field 分支），抽取完统一交给反序列化器。

分页接口怎么拉

分页集中在pageing块，两种类型。PageNumber（默认）：page_field指定请求里的分页字段名（默认page），start_page_number是起始页（默认 1），total_page_size = 0表示总页数未知——源码改用「单页返回行数 <pageing.batch_size」来终止（HttpSourceReader第 497 行readSize < pageInfo.getBatchSize()判断）；use_placeholder_replacement = true时支持"10${page}"这种带前后缀的替换。Cursor：page_type = "Cursor"，cursor_field指定请求里的游标字段名，cursor_response_field指定从响应哪个 JSONPath 读游标，游标为空或和上一轮相同时结束（noMoreElementFlag）。

心法一句话：先画出最终发出的 HTTP 请求长什么样。最绕的三条规则：

• GET：params一定拼进 URL 查询串；
• POST且keep_params_as_form = false：body 按 JSON 发送，没配 body 会发空 JSON{}；
• keep_params_as_form = true：params 并入表单 body，SeaTunnel 自动补application/x-www-form-urlencoded头；keep_page_param_as_http_param = true时分页字段直接写进params。

写入模式与 Upsert 机制

先记住一点：SeaTunnel 只有拿到主键 / 唯一键信息时才会进 upsert / update 路径。这个 key 可以来自显式配置的primary_keys；没配时尝试从上游 Catalog 元数据继承主键，再尝试第一组 unique key；都没有就退化成普通 INSERT。存在 key 且enable_upsert = true（默认true）时优先走方言原生语句，PostgreSQL 生成INSERT ... ON CONFLICT (...) DO UPDATE（所有字段都是主键时退化为DO NOTHING，见PostgresDialect）。任务里没有重复 key 数据时，把enable_upsert设为false，省掉冲突处理、加快导入。

⚡ 性能与可靠性旋钮

• batch_size：越大 flush 越少、吞吐越高，代价是内存占用与故障重试量同步变大；
• JDBC URL 加rewriteBatchedStatements=true：MySQL 非 XA 批量任务把语句重写成多值 INSERT，吞吐立竿见影；
• use_copy_statement = true：PostgreSQL 走COPY FROM STDIN路径，要求驱动提供getCopyAPI()，且不支持MAP/ARRAY/ROW类型；
• is_exactly_once = true+xa_data_source_class_name+max_retries = 0：XA 两阶段提交，数据库和驱动都要支持 XA（PostgreSQL 需开 prepared transaction），开之前务必确认前置条件；
• batch_interval_ms：大于 0 时按时间触发 flush，低流速场景用它控制端到端延迟。

扩展方向

链路通了之后，复用只是换参数的事：把 sink 的url和driver换成 MySQL / Oracle / SQL Server（驱动参考见 JDBC Sink 文档），把 source 的url和headers换成带鉴权的真实接口；想上 Flink / Spark 引擎，配置基本不用动，注意驱动要放到每个节点的plugins/Jdbc/lib/下。上面这份配置可以直接当「拉 API 数据入库」这类场景的模板用。

延伸资源

• Http Source 连接器文档
• JDBC Sink 连接器文档
• 部署与连接器插件安装
• e2e 参考配置：http_streaming_json_to_postgresql.conf
• 源码：HttpSourceReader.java
• 源码：JdbcSinkFactory.java

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel