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

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 postgresql4. 先看 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 trueSeaTunnel 会自己发 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?loggerLevelOFF 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枚举定义在HttpSourceOptionsformat声明响应体类型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_ordersprimary_keys既参与建表主键约束也参与 Upsertbatch_size是攒满多少行触发一次 flushsink 侧默认 1000示例里设了 100。两个 SaveMode 参数都有默认值通常可省schema_save_mode默认CREATE_SCHEMA_WHEN_NOT_EXIST表不存在则创建另有RECREATE_SCHEMA、ERROR_WHEN_SCHEMA_NOT_EXIST、IGNOREdata_save_mode默认APPEND_DATA追加另有DROP_DATA清空后写、CUSTOM_PROCESSING配合custom_sql预处理、ERROR_WHEN_DATA_EXISTS有数据即报错。执行环境parallelism 1不是保守是上限Http Source 是单分片 readerHttpSourceReader继承AbstractSingleSplitReader水平拆不开调高并发也不会拉得更快。真正的选择是job.modeBATCH拉完即退出接口是增量返回、要定时轮询时切STREAMING并在 source 上配poll_interval_millise2e 配置 http_streaming_json_to_postgresql.conf 就是这个用法5 秒一轮。 核对结果 排错决策树任务结束后跑两条 SQLSELECT 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_fieldschema 在顶层空转任务成功但行数明显偏少源接口有分页作业只取了第一页没配pageing块配了total_page_size的数值是否小于实际总页数自动建表成功但写入反复失败、或数据被莫名覆盖primary_keys选的列能否唯一标识一行键选错upsert 时要么主键冲突要么静默覆盖数据从此不符合预期HTTP 状态码非 2xxurl、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()里的getPartOfJsoncontent_field 分支与decodeJSONparseToMapjson_field 分支抽取完统一交给反序列化器。分页接口怎么拉分页集中在pageing块两种类型。PageNumber默认page_field指定请求里的分页字段名默认pagestart_page_number是起始页默认 1total_page_size 0表示总页数未知——源码改用「单页返回行数 pageing.batch_size」来终止HttpSourceReader第 497 行readSize pageInfo.getBatchSize()判断use_placeholder_replacement true时支持10${page}这种带前后缀的替换。Cursorpage_type Cursorcursor_field指定请求里的游标字段名cursor_response_field指定从响应哪个 JSONPath 读游标游标为空或和上一轮相同时结束noMoreElementFlag。心法一句话先画出最终发出的 HTTP 请求长什么样。最绕的三条规则GETparams一定拼进 URL 查询串POST且keep_params_as_form falsebody 按 JSON 发送没配 body 会发空 JSON{}keep_params_as_form trueparams 并入表单 bodySeaTunnel 自动补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 加rewriteBatchedStatementstrueMySQL 非 XA 批量任务把语句重写成多值 INSERT吞吐立竿见影use_copy_statement truePostgreSQL 走COPY FROM STDIN路径要求驱动提供getCopyAPI()且不支持MAP/ARRAY/ROW类型is_exactly_once truexa_data_source_class_namemax_retries 0XA 两阶段提交数据库和驱动都要支持 XAPostgreSQL 需开 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考