企业官网建设流程全解析

SeaTunnel Http Source 连接器完整指南：HTTP 数据采集、分页抓取、JSON 抽取与二进制下载

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

Http Source 是 SeaTunnel 中用于从 HTTP API 读取数据的源连接器，支持 GET/POST 请求、JSON/text/binary 三种响应格式、基于页码与游标的分页抓取，以及流式轮询，是打通外部 HTTP 服务与 SeaTunnel 数据集成管道的最常用入口。阅读本文后，你将掌握 Http Source 的全部配置参数、分页与占位符替换机制、content_field/json_field的 JSON 抽取技巧，以及底层源码实现原理，能够独立完成从 HTTP 接口同步数据到各类 Sink 的实战作业。

Http 源连接器概述

Http 源连接器用于从 HTTP 接口读取数据，在 SeaTunnel 中通过Http插件标识注册（对应源码 HttpSourceFactory.java 中的factoryIdentifier()返回值）。

支持的引擎

Spark
Flink
SeaTunnel Zeta

主要特性

特性说明详见 Connector V2 特性文档。

依赖安装

源选项速查表

以下为 Http 源连接器的完整源选项，其中部分选项在源码 HttpSourceOptions.java 与 HttpCommonOptions.java 中定义：

注意：json_filed_missed_return_null沿用了源码中的拼写（Filed而非Field），作业配置中需保持该拼写，对应 HttpSourceOptions.java 中的JSON_FILED_MISSED_RETURN_NULL定义。

参数说明与易错点

• pageing是连接器现有的参数名，作业配置中必须保持这个拼写，不要写成paging。
• format = json通常需要配置schema。当字段来自嵌套 JSON 路径时，可以配合json_field使用；如果想直接抽取某个 JSON 数组或对象片段，可以使用content_field。HttpSourceFactory的optionRule()中通过conditional规则将schema声明为format = json时的条件必填项。
• format = binary会输出固定字段(data: bytes, relativePath: string, partIndex: long)，只支持批模式，且不能与pageing一起使用——源码 HttpSourceReader.java 的构造函数会在binaryMode && pageInfo != null时直接抛出CONFIG_VALIDATION_FAILED异常。

快速上手：创建 Http 数据同步作业

以下是一个完整的批处理作业：从 HTTP 接口读取 JSON 数据并按声明 schema 解析，最后输出到控制台：

env { parallelism = 1 job.mode = "BATCH" } source { Http { plugin_output = "http" url = "http://mockserver:1080/example/http" method = "GET" format = "json" schema = { fields { c_map = "map<string, string>" c_array = "array<int>" c_string = string c_boolean = boolean c_tinyint = tinyint c_smallint = smallint c_int = int c_bigint = bigint c_float = float c_double = double c_bytes = bytes c_date = date c_decimal = "decimal(38, 18)" c_timestamp = timestamp c_row = { C_MAP = "map<string, string>" C_ARRAY = "array<int>" C_STRING = string C_BOOLEAN = boolean C_TINYINT = tinyint C_SMALLINT = smallint C_INT = int C_BIGINT = bigint C_FLOAT = float C_DOUBLE = double C_BYTES = bytes C_DATE = date C_DECIMAL = "decimal(38, 18)" C_TIMESTAMP = timestamp } } } } } # 控制台打印读取的 Http 数据 sink { Console { parallelism = 1 } }

从源码 HttpClientProvider.java 的实现看，连接器使用 Apache HttpClient 执行请求，并将connect_timeout_ms、socket_timeout_ms注入RequestConfig。响应状态码在 200～207 之间视为成功，否则抛出REQUEST_FAILED异常。

format 详解：json / text / binary

format = json

当指定format为json时，应同时指定schema选项。例如上游数据如下：

{ "code": 200, "data": "get success", "success": true }

对应的 schema 配置：

schema { fields { code = int data = string success = boolean } }

连接器将生成如下数据：

format = text

当指定format为text时，连接器不会对上游数据做任何处理，整段响应文本作为一行content输出。同样输入上述 JSON，生成的数据为：

配合enable_multi_lines = true时，HttpSourceReader.pollAndCollectData 会将响应文本按行拆分为多条记录。

format = binary

当指定format为binary时，HTTP 响应体作为原始字节处理，用于下载文件（PDF、图片、ZIP 等）。输出 schema 固定为(data: bytes, relativePath: string, partIndex: long)。大文件会根据binary_chunk_size自动拆分为多行，文件名优先从响应头Content-Disposition提取（对应源码 FilenameExtractor.java）。仅支持 BATCH 模式，且不支持分页（pageing）。

示例：通过 HTTP 下载文件并写入 LocalFileSink：

env { parallelism = 1 job.mode = "BATCH" } source { Http { url = "http://example.com/files/report.pdf" method = "GET" format = "binary" binary_chunk_size = 10485760 # 每个分片 10MB schema = { fields { data = bytes relativePath = string partIndex = long } } } } sink { LocalFile { path = "/tmp/download" file_format = "binary" } }

二进制模式在底层通过 HttpClientProvider.executeBinaryStreaming 以流式分块方式消费响应体，每读到chunkSize字节即回调一次，避免整个大文件驻留内存。

请求组装细节：params、body、headers 与兼容开关

keep_params_as_form

用于兼容旧版本的 http 连接器行为：

• 当设置为true时，params和pageing以表单形式提交。
• 当设置为false时，params添加到 URL 路径中，而pageing不会添加到 body 或表单中，只会替换 params 和 body 中的占位符。

从源码 HttpClientProvider.execute 看：POST +keepParamsAsForm = true时，若params非空，会自动补上Content-Type: application/x-www-form-urlencoded，并将body与params合并为表单 body；若两者出现同名键，params的值会覆盖body中的同名键。

keep_page_param_as_http_param

控制分页参数是否写入params：

• 当设置为true时，pageing写入params。
• 当设置为false时，当页面字段已存在于body或params中时替换其值。

当设置为false时，配置示例：

body="""{"id":1,"page":"${page}"}"""

params={ page: "${page}" }

对应 HttpSourceReader.updateRequestParam：keepPageParamAsHttpParam = true时直接向 params 写入分页字段并立即返回；否则按“headers → params → body”的顺序，仅更新已存在的同名键或占位符，不会凭空新增分页字段。

params

默认情况下，参数将添加到 URL 路径（查询串）中。如果需要保持旧版本行为，请参考keep_params_as_form。

body

HTTP body 用于在请求或响应中携带实际数据，包括 JSON、表单提交。参考格式：

body="""{"id":1,"name":"seatunnel"}"""

对于表单提交，请按如下设置 content-type：

headers { Content-Type = "application/x-www-form-urlencoded" }

源码中有一个值得注意的实现细节：对于 POST 请求，若未开启keepParamsAsForm且未显式将Content-Type设为 form 类型，body 会以原始字符串（StringEntity，Content-Type: application/json）直接发送，从而保留嵌套 JSON 结构（如data.type不会被展平）；仅当显式声明 form Content-Type 或开启keepParamsAsForm时，body 才会走 HOCON 解析并转为表单参数。

分页与最终请求形态排查

下面几条规则最容易混淆，建议先按“最终发出的 HTTP 请求长什么样”来理解：

1. GET请求：params一定会被拼到 URL 查询串里。
2. POST且keep_params_as_form = false：
   • params仍然会拼到 URL 查询串里；
   • 在默认的非 form 分支上，body会作为 JSON body 发送；
   • 如果没有配置body，并且请求仍然走这个默认非 form 分支，运行时会发送一个空 JSON 对象{}作为请求体；
   • 如果显式把Content-Type设为application/x-www-form-urlencoded，运行时会改走 form-body 分支，而不是默认 JSON 分支。
3. POST且keep_params_as_form = true：
   • params会并入表单 body；
   • 如果未显式设置Content-Type，SeaTunnel 会自动补application/x-www-form-urlencoded；
   • 如果body与params出现同名键，params的值会覆盖body中同名键。
4. keep_page_param_as_http_param = true：分页字段会直接写入params。
5. keep_page_param_as_http_param = false：SeaTunnel 只会更新 headers、params、body 里已经存在的同名键或占位符，不会凭空新增分页字段。
6. pageing.use_placeholder_replacement = true：支持${page}、${cursor}占位符，也支持"10${page}" -> "105"这种带前后缀的替换；为false时只做按 key 的整值替换（对应源码replacePlaceholder与processPageMap的两条分支）。

示例 1：GET 分页，页码写入查询参数

source { Http { url = "https://api.example.com/orders" method = "GET" params = { page = "${page}" size = "100" } pageing = { page_field = "page" page_type = "PageNumber" start_page_number = 3 use_placeholder_replacement = true } } }

当页码推进到3时，最终请求为：

GET https://api.example.com/orders?page=3&size=100

示例 2：POST JSON（默认非 form 分支），请求参数进 URL，分页字段留在 body

source { Http { url = "https://api.example.com/orders/search" method = "POST" keep_params_as_form = false params = { tenant = "acme" } body = """{"page":"${page}","pageSize":100}""" pageing = { page_field = "page" page_type = "PageNumber" start_page_number = 3 use_placeholder_replacement = true } } }

当页码推进到3时，最终请求为：

POST https://api.example.com/orders/search?tenant=acme Content-Type: application/json Body: {"page":"3","pageSize":100}

示例 3：POST 表单，请求参数和分页字段都写入表单 body

source { Http { url = "https://api.example.com/orders/search" method = "POST" keep_params_as_form = true keep_page_param_as_http_param = true params = { size = "100" } pageing = { page_field = "page" page_type = "PageNumber" start_page_number = 3 } } }

当页码推进到3时，最终请求为：

POST https://api.example.com/orders/search Content-Type: application/x-www-form-urlencoded Body: size=100&page=3

content_field：直接抽取 JSON 片段

content_field可以获取部分 JSON 数据。如果只需要book部分的数据，配置content_field = "$.store.book.*"。假设返回数据如下：

{ "store": { "book": [ { "category": "reference", "author": "Nigel Rees", "title": "Sayings of the Century", "price": 8.95 }, { "category": "fiction", "author": "Evelyn Waugh", "title": "Sword of Honour", "price": 12.99 } ], "bicycle": { "color": "red", "price": 19.95 } }, "expensive": 10 }

配置content_field = "$.store.book.*"后返回结果：

[ { "category": "reference", "author": "Nigel Rees", "title": "Sayings of the Century", "price": 8.95 }, { "category": "fiction", "author": "Evelyn Waugh", "title": "Sword of Honour", "price": 12.99 } ]

然后可以使用更简单的 schema 获取所需结果：

Http { url = "http://mockserver:1080/contentjson/mock" method = "GET" format = "json" content_field = "$.store.book.*" schema = { fields { category = string author = string title = string price = string } } }

参考实现：

• 测试数据：mockserver-config.json
• 任务配置：http_contentjson_to_assert.conf

实现原理上，HttpSourceReader.collect 通过JsonPath.compile(contentJson)编译 JSONPath 表达式并截取响应片段，再交给下游 deserialization 处理。

json_field：按 JSONPath 映射字段

json_field用于辅助配置 schema，因此必须与schema一起使用。对于上面的同一份数据，可以通过如下配置获取book的内容：

source { Http { url = "http://mockserver:1080/jsonpath/mock" method = "GET" format = "json" json_field = { category = "$.store.book[*].category" author = "$.store.book[*].author" title = "$.store.book[*].title" price = "$.store.book[*].price" } schema = { fields { category = string author = string title = string price = string } } } }

参考实现：

• 测试数据：mockserver-config.json
• 任务配置：http_jsonpath_to_assert.conf

json_field与content_field的区别在于：content_field把整个数组片段交给 schema 逐字段解析；而json_field为每个字段单独指定 JSONPath，连接器在 JsonPathProcessorImpl.java / ArrayJsonPathProcessor.java 中按路径取值并做行列转置（dataFlip）后映射成行。当 JSON 字段缺失时，json_filed_missed_return_null = true会返回 null 而非报错。

pageing 分页详解：PageNumber 与 Cursor

当前支持的分页类型是PageNumber和Cursor（定义见 HttpPaginationType.java）。默认分页类型为PageNumber。分页信息载体为 PageInfo.java，包含totalPageSize、batchSize、pageField、pageIndex、pageType、cursor、pageCursorFieldName、pageCursorResponseField、usePlaceholderReplacement等字段。

1. PageNumber

使用PageNumber分页时，页面参数可放在 HTTP 请求的不同部分：

• 在 URL 参数中：将页面参数添加到params部分；
• 在请求体中：在bodyJSON 中包含页面参数；
• 在头信息中：将页面参数添加到headers部分。

可使用${page}占位符配合use_placeholder_replacement = true动态更新这些值，占位符支持多种格式：

• 作为独立值："${page}"
• 带前缀/后缀："10${page}"或"page-${page}"
• 作为不带引号的数字：${page}（在 JSON 体中）
• 在嵌套 JSON 结构中：{"pagination":{"page":${page}}}

示例 1：在 body 和 params 中使用页面参数

source { Http { url = "http://localhost:8080/mock/queryData" method = "POST" format = "json" body="""{"id":1,"page":"${page}"}""" content_field = "$.data.*" params={ page: "${page}" } pageing={ #你可以不设置此参数，默认值是 PageNumber page_type="PageNumber" total_page_size=20 page_field=page use_placeholder_replacement=true #当不知道 total_page_size 时使用 batch_size，如果读取大小<batch_size 则完成，否则继续 #batch_size=10 } schema = { fields { name = string age = string } } } }

示例 2：在 headers 中使用页面参数

source { Http { url = "http://localhost:8080/mock/queryData" method = "GET" format = "json" headers={ Page-Number = "${pageNo}" Authorization = "Bearer token-123" } pageing={ page_field = pageNo start_page_number = 1 batch_size = 10 use_placeholder_replacement = true } schema = { fields { name = string age = string } } } }

示例 3：使用基于键的替换（不使用占位符）

source { Http { url = "http://localhost:8080/mock/queryData" method = "GET" format = "json" params={ page = "1" } pageing={ page_field = page start_page_number = 1 batch_size = 10 use_placeholder_replacement = false } schema = { fields { name = string age = string } } } }

示例 4：在 headers 中使用带前缀的页码

source { Http { url = "http://localhost:8080/mock/queryData" method = "GET" format = "json" headers = { Page-Number = "10${page}" # 当 page=5 时将变为 "105" Authorization = "Bearer token-123" } pageing = { page_field = page start_page_number = 5 batch_size = 10 use_placeholder_replacement = true } schema = { fields { name = string age = string } } } }

示例 5：在 body 中使用不带引号的页码

source { Http { url = "http://localhost:8080/mock/queryData" method = "POST" format = "json" body = """{"a":${page},"limit":10}""" # 不带引号的数字 pageing = { page_field = page start_page_number = 1 batch_size = 10 use_placeholder_replacement = true } schema = { fields { name = string age = string } } } }

示例 6：使用带页面参数的嵌套 JSON 结构

source { Http { url = "http://localhost:8080/mock/queryData" method = "POST" format = "json" body = """{"pagination":{"page":${page},"size":10},"filters":{"active":true}}""" # 嵌套结构 pageing = { page_field = page start_page_number = 1 total_page_size = 20 use_placeholder_replacement = true } schema = { fields { name = string age = string } } } }

2. Cursor（游标分页）

pageing.page_type必须设置为Cursor；cursor_field是请求参数中游标的字段名称；cursor_response_field是响应数据中分页令牌字段的名称（支持 JSONPath），连接器会将其值写入后续请求的分页字段中。

source { Http { plugin_output = "http" url = "http://localhost:8080/mock/cursor_data" method = "GET" format = "json" content_field = "$.data.*" keep_page_param_as_http_param = true pageing ={ page_type="Cursor" cursor_field ="cursor" cursor_response_field="$.paging.cursors.next" } schema = { fields { content=string id=int name=string } } json_field = { content = "$.data[*].content" id = "$.data[*].id" name = "$.data[*].name" } } }

游标分页的停止条件在 HttpSourceReader.collect 中实现：从响应 JSON 中读取cursor_response_field对应的首个值作为新游标；当新游标为空或与当前游标相同（无法推进）时停止翻页。实际可运行的分页用例可参考 e2e 测试资源，例如 http_page_increase_page_num.conf、http_page_increase_no_page_num.conf、http_page_cursor_num_assert.conf。

源码级原理：连接器内部如何工作

参数模型

所有源选项以Option常量形式集中在 HttpSourceOptions.java（源专属）与 HttpCommonOptions.java（url/headers/params/retry 系列，与 Http Sink 共用）中，再由 HttpSourceFactory.java 组装成OptionRule完成参数校验（url必填、format=json时schema条件必填）。

读取主循环

HttpSourceReader.internalPollNext 是核心循环：

• 未配置pageing时，单次请求后结束（批模式）或按poll_interval_millis睡眠后继续（流模式）；
• PageNumber 分页时，每轮updateRequestParam更新请求参数，pollAndCollectData拉取数据后页码自增 1；
• 结束条件两条路径：配置了total_page_size > 0时，pageIndex >= totalPageSize即停止；未配置时，若当页读取行数< batch_size则停止（否则继续翻页）；
• 批模式下noMoreElementFlag为 true 时调用context.signalNoMoreElement()结束作业。

请求执行与重试

HttpClientProvider 基于 Apache HttpClient 实现：GET通过URIBuilder拼接查询参数；POST依据keepParamsAsForm与 Content-Type 在 JSON body 与表单 body 之间切换；重试使用guava-retrying的RetryerBuilder，仅对IOException生效，退避策略为斐波那契退避，下限retry_backoff_multiplier_ms（默认 100ms）、上限retry_backoff_max_ms（默认 10s），停止策略为retry指定的最大尝试次数。

参考资料

• 连接器变更日志：connector-http.md
• e2e 测试用例目录：connector-http-e2e/src/test/resources，内含http_json_to_assert.conf、http_binary_to_assert.conf、http_multilinejson_to_assert.conf、http_formrequestbody_to_assert.conf、http_streaming_json_to_postgresql.conf等覆盖各类格式与分页场景的作业示例
• 源码位置：connector-http-base

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