企业官网建设流程全解析

用 SeaTunnel 把 HTTP API 数据写入 PostgreSQL：从接口到落库的完整实战指南

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

什么场景需要这条链路？

如果你的数据源头是一个返回 JSON 的 HTTP API（监控接口、业务报表接口、第三方数据接口），目标是 PostgreSQL、MySQL 这类关系型数据库，而且你不想自己写定时脚本加 ETL 代码，那这篇就是为你准备的。用 SeaTunnel 的 Http Source 拉取接口数据，再用 JDBC Sink 写入数据库，整条链路就是一个 Source 加一个 Sink：

HTTP API（JSON）──> Http Source ──> SeaTunnel Row ──> JDBC Sink ──> PostgreSQL 表

整篇按「先搭环境、再写配置、后跑验收」的顺序展开，最后附排错地图和参数速查表。

两个连接器各自负责什么？

把整条链路想成一次交接：Http Source 像翻译官，把接口的 JSON 响应翻译成标准的数据行（SeaTunnelRow）；JDBC Sink 像快递员，把这些行按批次送进数据库表里。翻译官只负责"读懂并结构化"，快递员只负责"送达并写对表"，中间流动的 SeaTunnelRow 就是两者之间的标准信封。

• Http Source：按url、method（仅支持GET/POST）发起请求，拿到响应后按你声明的schema把 JSON 反序列化成结构化行。它不支持用户自定义分片，单并发的接口拉取是它的舒适区。
• JDBC Sink：通过数据库厂商提供的 JDBC 驱动写库，支持自动生成 INSERT / 原生 UPSERT / UPDATE / DELETE、自动建表，以及基于primary_keys的主键 Upsert。

环境准备清单：动手前先验证一遍

把下面五个检查点全部打勾，可以省掉后面 90% 的排查时间：

• 跑通第一个任务：先用config/v2.batch.config.template验证本地能启动 SeaTunnel、配置能解析、执行引擎正常（对应官方「跑第一个任务」教程）。
• 安装链路插件：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)'

插件名与 JAR 的对应关系可在${SEATUNNEL_HOME}/connectors/plugins-mapping.properties中查询（仓库根目录的 plugin-mapping.properties 是同源映射）。

• 放好 JDBC 驱动：Zeta 引擎（本地模式）把postgresql-42.x.x.jar放进${SEATUNNEL_HOME}/lib后重启受影响的 SeaTunnel 进程；Spark / Flink 引擎则放到每个执行节点的${SEATUNNEL_HOME}/plugins/Jdbc/lib/。驱动不统一内置，是因为各厂商许可证与再分发条款不同，且版本必须同时兼容目标库与 Java 运行时。

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

• 先 curl 看一眼返回内容：确认 JSON 顶层就有你要的字段。本文示例接口返回的顶层包含c_string、c_int、c_bigint等字段（mock 定义见 mockserver-config.json）。如果你的真实接口把有效数据包在{"code":200,"data":{...}}这类深层结构里，先停下来，后面要补字段抽取配置。

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

• 准备 PostgreSQL 目标库并授权：因为本教程用generate_sink_sql = true自动建表，sink 用户必须在publicschema 有建表权限：

CREATE USER test WITH PASSWORD 'test'; CREATE DATABASE test OWNER test; -- 重新连接到 test 库后执行 GRANT USAGE, CREATE ON SCHEMA public TO test;

配置解剖：按数据流的三个环节写配置

把配置保存为config/http-to-jdbc.conf。下面不按 env/source/sink 分块讲，而是跟着数据走一遍：请求怎么发出去 → 响应怎么解析 → 行怎么落库。

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 } }

第一环节：请求怎么发出去

• url+method：决定打到哪个接口、用什么方法。method只认GET和POST，写成别的会直接报错。
• plugin_output = "http_orders"：给本节点的输出流起名，sink 靠plugin_input同名对接。名字写不一致，上下游就接不上。
• parallelism = 1：Http Source 不支持自定义分片，单页接口单并发即可，写大也没有意义。
• job.mode = "BATCH"：批模式，拉完即退出；想定时轮询接口就改成STREAMING并配poll_interval_millis（仓库 e2e 用例 http_streaming_json_to_postgresql.conf 就是 5 秒轮询一次）。

第二环节：响应怎么解析

• format = "json"：上游数据格式，支持json、text、binary，默认是text。设为json后schema是必填的——它是反序列化的唯一依据。
• schema.fields：声明字段名和类型（如c_string = string、c_int = int）。写错的后果很直接：字段名对不上或类型不匹配，解析阶段就会失败，整批数据出不来。

第三环节：行怎么落库

• generate_sink_sql = true是写入模式开关，默认false。为true时由 SeaTunnel 根据上游 schema 自动生成 SQL 和建表语句，此时database必填（本文写test）；为false时你必须自己提供query自定义 SQL，且此模式不执行 SaveMode。两种模式互斥，不可混用——没有显式写generate_sink_sql = true却忘了query，任务会在参数校验阶段就失败。
• table = "public.http_orders"：有 schema 概念的库必须写成schema.table形式，漏写 schema 会建到错误位置或报错。
• primary_keys = ["c_string"]：这个键有两个用途——生成原生 UPSERT/UPDATE/DELETE，以及参与自动建表时的主键约束。选错键的代价是重复写入时主键冲突或静默覆盖。
• batch_size = 100：每个批次最多缓存的行数，攒满才 flush 一次。调大减少写库次数、提升吞吐，代价是内存占用和失败重试量同步上涨。

运行与验收：一条命令加两条断言

启动命令：

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

-m local表示本地模式运行，不启动独立引擎集群，适合单机验证。

成功标准不是"进程没报错"，而是以下三条断言全部成立：

1. 日志断言：任务日志中没有任何 HTTP 解析错误（schema 解析失败的典型信号）和 JDBC DDL 错误（自动建表失败的典型信号）。
2. 行数断言：

SELECT COUNT(*) FROM public.http_orders;

结果应等于本次接口返回的行数（mock 接口为单页单行时就是 1）。

3. 内容断言：

SELECT c_string, c_int FROM public.http_orders ORDER BY c_string;

查询出的值应与curl看到的接口返回一致——mock 数据的c_string是形如"WArEB"的随机字符串，能对上就说明"翻译官"和"快递员"都没掉包。

排错地图：症状、原因与动作

按实际踩坑概率从高到低排序：

进阶旋钮：按需打开的开关

以下能力默认都是关闭或简化的，遇到对应场景再打开。

分页拉取（pageing）——接口返回数据超过一页时打开。PageNumber是默认类型：page_field指定请求里的分页字段（默认page），start_page_number默认 1；total_page_size填 0 表示总页数未知，此时连接器在单页返回行数小于pageing.batch_size（默认 100）时自动停止。设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头。

嵌套字段抽取——返回体把有效数据藏在深层时打开。content_field直接抽取整个片段，如content_field = "$.store.book.*"后 schema 只需声明book元素内的字段；json_field则为每个 schema 字段单独指定 JSONPath（如{ category = "$.store.book[*].category" }），必须与schema一起使用。

Upsert——接口重复推送同一批 key 时打开。SeaTunnel 只有拿到主键/唯一键信息才走 upsert 路径：优先用显式primary_keys，没有则尝试从上游 Catalog 元数据继承，再退到第一组 unique key，都没有就退化为普通 INSERT。有键且enable_upsert = true（默认）时优先使用方言原生语句，PostgreSQL 会生成INSERT ... ON CONFLICT (...) DO UPDATE（所有字段都是主键时退化为DO NOTHING）。数据无重复 key 时可设enable_upsert = false加快导入。

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。两者都有默认值，大多数场景可省略。

吞吐与一致性——非 XA 的 MySQL 批量任务在 JDBC URL 加rewriteBatchedStatements=true；PostgreSQL 大批量导入可试use_copy_statement = true走COPY路径（要求驱动提供getCopyAPI()，且不支持MAP/ARRAY/ROW类型）。要精确一次，配is_exactly_once = true+xa_data_source_class_name+max_retries = 0，前提是数据库与驱动都支持 XA（PostgreSQL 需启用 prepared transaction）。

一页速查：核心参数速查表

延伸阅读

• Http Source 连接器文档：全部源选项、format 详解、分页示例
• JDBC Sink 连接器文档：写入模式、SaveMode、exactly-once、故障排查与驱动参考
• 官方食谱：HTTP 到 JDBC
• 部署与下载连接器插件、跑第一个任务
• e2e 参考实现：http_streaming_json_to_postgresql.conf（流式轮询 Http 接口并写入 PostgreSQL）
• 源码位置：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