企业官网建设流程全解析

从零搭建 Http Source + JDBC Sink 数据同步链路

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

假设运营后台每天要把第三方 HTTP API 的订单数据灌进 PostgreSQL 数仓。下面用 Apache SeaTunnel 的 Http Source + JDBC Sink 把这条数据同步链路搭通：插件安装、配置装配、写入验证，全程基于 Zeta 引擎本地模式，照做即可复现。

核心原理：两个连接器各干了什么

数据流经四个环节：

HTTP API（JSON 响应） │ GET / POST 请求 ▼ Http Source：按 schema 拉取并反序列化 → SeaTunnelRow │ plugin_output = "http_orders" ▼ JDBC Sink：按 schema 与 RowKind 生成 INSERT / 原生 UPSERT → 批量落库 │ batch flush ▼ PostgreSQL 目标表

为什么选这条链路？对比自己写 cron 脚本拉取再逐条插入，你不需要自己维护解析、重试与批量刷盘逻辑，连接器层已经处理；对比 DataX，配置是单份 HOCON 文件、插件按需安装，新增数据库只需补一个驱动 JAR。源码层面，HttpSourceReader在pollAndCollectData中拿到响应后交给DeserializationCollector，按你声明的 schema 产出结构化行；sink 侧按plugin_input接收这些行并写库。

环境准备清单：插件安装与驱动放置五项检查

动手前先走完以下五项，每项给出「怎么做 + 怎么确认完成」。

• 运行环境：SeaTunnel 已能正常启动并解析配置，完成跑第一个任务即可。
• 插件安装：2.2.0-beta 起发行包不再默认附带全部连接器，先把config/plugin_config收敛为：

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

然后执行并确认 JAR 落盘：

cd "${SEATUNNEL_HOME}" sh bin/install-plugin.sh ls connectors | rg 'connector-(http-base|jdbc)' # 两个 JAR 都出现即完成

插件名与 JAR 的对应关系可查plugin-mapping.properties，指定版本等细节见部署文档。

• 驱动放置：Zeta 本地模式把 PostgreSQL JDBC 驱动放入${SEATUNNEL_HOME}/lib并重启受影响的 SeaTunnel 进程；Spark / Flink 引擎则放入各执行节点的${SEATUNNEL_HOME}/plugins/Jdbc/lib/：

ls "${SEATUNNEL_HOME}/lib" | rg 'postgresql' # 出现 postgresql-42.x.x.jar 即完成

• 接口预检：先看接口实际返回什么：

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

该 mock 数据定义在 mockserver-config.json，JSON 顶层有c_string、c_int等字段。如果你的真实接口把有效数据包在更深层（如{"code":200,"data":{...}}），配置时就要补content_field或json_field。

• 目标库权限：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;

配置逐层装配：从最简骨架到完整配置

配置分四层装配，每层单独可运行，出错时更容易定位问题。

第一层：运行环境 + 源端必填。format = json必须配套schema，这是源端唯一硬性组合：

env { parallelism = 1 # Http Source 暂不支持自定义分片，示例接口单页返回 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 连接信息。缺任何一项 sink 都建立不了连接：

sink { Jdbc { plugin_input = "http_orders" driver = "org.postgresql.Driver" url = "jdbc:postgresql://postgresql:5432/test?loggerLevel=OFF" username = "test" password = "test" database = "test" # generate_sink_sql = true 时必填 table = "public.http_orders" # 有 schema 概念的库必须写 xxx.xxx } }

第三层：写入策略。这一步决定 DDL / DML 由 SeaTunnel 生成还是你自己提供，省掉它任务会卡在参数校验：

sink { Jdbc { # ... 上面的连接信息 generate_sink_sql = true primary_keys = ["c_string"] } }

第四层：性能参数：

sink { Jdbc { # ... batch_size = 100 # 每批最多缓存行数，达到后触发 flush } }

完整配置如下，可直接保存为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 } }

执行与结果确认：任务日志特征与验证 SQL

本地模式运行：

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

看日志下判断：如果看到 HTTP 解析异常，说明schema.fields的字段名或类型与接口返回不一致，先curl核对字段名；如果看到 JDBC DDL 权限报错，说明「目标库权限」那步的GRANT没做；如果日志干净走到任务结束，进入库内核对：

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

行数与接口返回条数一致、取值与curl输出相同，链路即通。

关键参数深潜：generate_sink_sql 写入模式选择

🔑 下面三个参数最易配错，统一按「语义 → 配错的典型表现 → 正确配法」三步讲。

generate_sink_sql 与 query 互斥

• 参数语义：generate_sink_sql = true时（定义于 JdbcSinkOptions.java，默认false），SeaTunnel 依据上游 schema 与 RowKind 自动生成 INSERT、原生 UPSERT、UPDATE、DELETE，并支持 SaveMode 与自动建表。
• 配错的典型表现：提交时报「缺少 query」类校验错误。在JdbcSinkFactory的校验逻辑中可以看到：url、driver、schema_save_mode、data_save_mode始终要求（后两者有默认值可省略）；generate_sink_sql = true时强制要求database，为false时强制要求query。
• 正确配法：两种写入模式二选一。自动模式配generate_sink_sql = true+database+table；自管 SQL 模式配query = "INSERT INTO ... VALUES (?, ...)"，?按上游字段顺序绑定，且该模式不执行 SaveMode。

primary_keys 决定 Upsert 路径

• 参数语义：primary_keys既用于生成数据库原生 UPSERT / UPDATE / DELETE，也参与自动建表时主键约束的生成。
• 配错的典型表现：选的键无法唯一标识记录，重复写入时出现主键冲突或静默覆盖；或者你以为在 Upsert，实际只是普通 INSERT。
• 正确配法：未配置primary_keys时，连接器先尝试从上游 Catalog 元数据继承主键，再取第一组 unique key，仍没有则退化为普通 INSERT。有键且enable_upsert = true（默认）时，PostgreSQL 生成INSERT ... ON CONFLICT (...) DO UPDATE，所有字段都是主键时退化为DO NOTHING。任务没有重复 key 数据时，可把enable_upsert设为false提速。

content_field 与 json_field 的适用边界

• 参数语义：返回体把有效数据包在深层时，content_field一次性抽出某个 JSON 数组或对象片段（如$.store.book.*），schema 只需声明片段内字段；json_field则为每个 schema 字段单独指定 JSONPath（如category = "$.store.book[*].category"），必须与schema一起使用。
• 配错的典型表现：两者都没配时，schema 落在顶层字段上取不到值；只配json_field却不配schema，校验直接报错。
• 正确配法：抽取逻辑在 HttpSourceReader.java 的collect()方法中——content_field走getPartOfJson分支，json_field走decodeJSON+parseToMap分支，再交给反序列化器。JSON 字段缺失默认抛错，可配json_filed_missed_return_null = true让缺失字段返回null。

实战变体：分页拉取、嵌套 JSON 与流式轮询

如果你的场景是接口有分页，在 Http Source 里加pageing块：

source { Http { # ... 上面的 source 配置 pageing { page_type = "PageNumber" # 默认按页码推进；游标接口用 Cursor page_field = "page" start_page_number = 1 total_page_size = 0 # 0 表示未知，单页行数 < pageing.batch_size（默认 100）时停止 } } }

注意：配置前先想清楚「最终发出的 HTTP 请求长什么样」——GET下params一定拼进 URL 查询串；use_placeholder_replacement = true时按${page}占位符替换（支持"10${page}"→"105"这类带前后缀形式）；游标接口改用page_type = "Cursor"，配cursor_field与cursor_response_field，响应游标为空或不变时结束拉取。

如果你的场景是有效数据包在深层字段，在 source 块二选一：

source { Http { # ... 上面的 source 配置 content_field = "$.store.book.*" # 或 json_field = { category = "$.store.book[*].category" } } }

注意：抽取后 schema 只声明片段内字段，且该参数的边界语义见上一节深潜，别把两种抽取方式混用。

如果你的场景是持续轮询接口写库（例如每 5 秒拉一次增量订单），切换流模式：

env { job.mode = "STREAMING" } source { Http { # ... 上面的 source 配置 poll_interval_millis = 5000 } }

注意：仓库 e2e 工程里有几乎一致的真实用例 http_streaming_json_to_postgresql.conf，流式写法可直接对照。

故障速查：Http Source 到 PostgreSQL 高频问题

⚠️ 表中没覆盖的症状，先curl核对接口返回——多数问题出在源端而非数据库端。

扩展方向：替换 Source 与 Sink 目标库

这条链路是「拉 API 数据入库」的通用模板，替换后无需改动结构。数据源变化时只改 source 块：把 Http Source 换成消息队列、文件类等其他连接器，参数随之调整，sink 配置原样保留；目标库变化时只改 Jdbc 块的driver、url与驱动 JAR（如 MySQL 为mysql-connector-j-8.x.x.jar、Oracle 为ojdbc8.jar），完整参数与驱动参考见 JDBC Sink 文档，更完整的流程对照可看官方食谱 http-to-jdbc。验证通过后把参数换成你业务真实的接口地址与表名即可直接投产。整条链路参数不多，却完整覆盖了 SeaTunnel「源端拉取反序列化 + 汇端 SQL 生成批量落库」的典型分工，PostgreSQL 跑通之后，同一套方法可以原样复制到你的其余目标库清单上。

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