企业官网建设流程全解析

SeaTunnel Http Source Connector 实战指南：配置全解、JSON 解析与分页拉取源码原理

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

Http Source 是 SeaTunnel 提供的 HTTP/HTTPS 数据源连接器，用于把任意 HTTP 接口的响应数据（JSON、文本或二进制文件）接入 SeaTunnel 数据管道，可直接在 Spark、Flink 与 SeaTunnel Zeta 引擎上以批处理或流处理方式运行。读完本文，你将掌握 Http Source 的全部 30+ 参数、format三种响应解析模式、content_field/json_field的 JSONPath 提取技巧、pageing页码与游标分页的完整用法，并能从源码层面理解请求构造、占位符替换与分页终止条件的确切行为，从而在真实业务中快速搭建 HTTP 数据同步任务并排查问题。

一、连接器能力边界与引擎支持

Http Source 在仓库中的实现位于 seatunnel-connectors-v2/connector-http/connector-http-base，其中Http为通用 HTTP 数据源，另有 airtable、feishu、github、gitlab、notion、shopify、stripe 等基于 HTTP 协议封装的专用连接器。

支持的引擎

特性支持矩阵

特性清单来自 connector-v2-features，勾选状态对应当前仓库实际能力：

其中「不支持 parallelism」的根因在源码中非常清晰：HttpSource继承自 AbstractSingleSplitSource（单分片 Source），Http 数据源天然只有一个数据分片，因此并行度恒为 1；流处理模式（job.mode = "STREAMING"）下配合poll_interval_millis轮询即可实现持续拉取。

依赖引入

使用该连接器需要引入connector-http依赖，可通过install-plugin.sh脚本安装，或从 Maven 中央仓库获取（坐标org.apache.seatunnel:connector-http）。依赖信息与安装方式可参考仓库 plugins/README.md 与 docs/en/developer/setup.md。

二、Source 参数全解

下表完整列出 Http Source 的全部参数，默认值出处为 HttpSourceOptions.java 与 HttpCommonOptions.java 中的Option定义。

关键参数源码级解读

pageing的拼写是刻意的：连接器选项名就是pageing（而非paging），配置作业时必须保持这一拼写。源码 HttpSourceOptions.java 中对应Options.key("pageing").mapType()。

total_page_size = 0的语义：从 HttpSourceReader.collect() 的实现看，当totalPageSize > 0时以「当前页号 >= 总页数」作为终止条件；否则以「单页实际返回行数 < batch_size」作为终止条件——即当返回行数小于batch_size时认为已拉取完毕，否则继续下一页。

format的取值来源：HttpConfig.ResponseFormat枚举定义了JSON("json")、TEXT("text")、BINARY("binary")三种取值，定义在 HttpConfig.java。

超时参数：连接超时与 Socket 超时在 HttpClientProvider 构造时通过 Apache HttpClient 的RequestConfig注入：setConnectTimeout(connectTimeoutMs)与setSocketTimeout(socketTimeoutMs)。

三、快速上手：创建 Http 数据同步任务

以下是一个完整可运行的批处理作业：从 MockServer 拉取 JSON 数据，按schema解析成结构化行，最终打印到 Console。该配置即原文档示例，同时与仓库 e2e 用例 http_json_to_assert.conf 使用的数据源（mockserver-config.json 中/example/http路径）保持一致：

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 } } } } } # Console printing of the read Http data sink { Console { parallelism = 1 } }

任务配置文件的书写格式遵循 config 文档，编写完成后用以下命令提交（以 SeaTunnel Zeta 引擎为例）：

# 先通过 install-plugin.sh 安装 connector-http 插件，再执行 bin/seatunnel.sh --config <your-http-job.conf> -e local

配置说明：schema中的字段类型即 SeaTunnel 类型系统支持的标量与复杂类型（含嵌套row），解析逻辑在 HttpSource.buildSchemaWithConfig() 中通过CatalogTableUtil.buildWithConfig构建表结构，format = json时使用JsonDeserializationSchema完成反序列化。

四、响应数据解析：format 详解

format = json

json格式要求配置schema，连接器将 JSON 响应按 schema 字段映射为结构化行。例如上游返回：

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

配置 schema：

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

连接器生成的数据如下：

从源码看，format = json时若同时配置了json_field或content_field，会先执行 JSONPath 提取再交给反序列化器；json_filed_missed_return_null = true时字段缺失返回 null 而非报错（详见第六节）。

format = text

text格式下连接器对上游数据不做解析，整个响应文本作为一列输出。上游数据为：

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

输出结果为单行单列：

此时无需schema，HttpSource 会为该模式生成一个固定的单字段表结构(content: string)，并使用SimpleTextDeserializationSchema反序列化。若响应是多行文本（如逐行 JSON），可开启enable_multi_lines = true，读取器在 HttpSourceReader.pollAndCollectData() 中会用BufferedReader按行切分后逐行产出。

format = binary（HTTP 文件下载）

binary格式把 HTTP 响应体当作原始字节流，用于下载 PDF、图片、ZIP 等文件。输出 schema 固定为三列：data: bytes、relativePath: string、partIndex: long。大文件会依据binary_chunk_size（默认 10MB）自动拆分为多行，partIndex标识块序号。该模式：

• 仅支持 BATCH 模式（HttpSource.getBoundedness() 中对 STREAMING 模式直接抛出HTTP binary format only supports BATCH mode异常）；
• 不能与pageing同时使用（HttpSourceReader 构造器中对binaryMode && pageInfo != null会抛出配置校验异常）；
• 文件名优先取响应头Content-Disposition，否则回退到 URL 中的文件名，逻辑见 FilenameExtractor.java。

文件下载并写入本地文件系统的示例（来自原文档，仓库另有对应 e2e 用例 http_binary_to_assert.conf）：

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

底层实现上，HttpClientProvider.executeBinaryStreaming() 以流式方式分块读取响应体（byte[] buffer = new byte[(int) chunkSize]），每读满一块就回调一次消费者产出(byte[] data, String filename, long partIndex)，避免整个文件驻留内存。

五、请求构造：params、body、headers 与兼容开关

params

默认情况下，params会被拼接到 URL 查询串上（URIBuilder.setParameter）。若需要保留旧版本行为（以表单提交），请设置keep_params_as_form = true。

body

body用于携带请求体（JSON 或表单内容）。参考写法：

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

表单提交时需要显式指定 Content-Type：

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

源码中 HttpClientProvider.addBody() 会依据请求头是否为application/x-www-form-urlencoded决定走表单编码（UrlEncodedFormEntity）还是 JSON 实体（StringEntity+application/json），若用户未定义 Content-Type 则自动补application/json。

keep_params_as_form

用于兼容旧版 Http 连接器行为：

• 为true时：params与pageing以表单方式提交；若未显式设置 Content-Type，SeaTunnel 会自动加上application/x-www-form-urlencoded；当body与params含相同 key 时，params的值覆盖body的值。
• 为false时：params追加到 URL 路径；pageing不写入 body 或 form，而是替换 params 与 body 中的占位符。

keep_page_param_as_http_param

• 为true时：pageing字段直接写入params（对应 HttpSourceReader.updateRequestParam() 中keepPageParamAsHttpParam分支，将 page/cursor 放入 params）。
• 为false时：仅更新 body 或 params 中已存在的 key 或占位符，不会自动发明新的分页字段。

为false时的配置示例：

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

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

分页与最终请求形态速查

排查 Http Source 问题最有效的方式是从最终发出的请求反推配置。总结规则如下：

1. 对于GET，params总是追加到 URL 查询串。
2. 对于POST+keep_params_as_form = false：
   • params仍进入 URL 查询串；
   • 默认非表单路径下，body作为 JSON 请求体序列化；
   • 若未配置body且保持默认非表单路径，运行时发送空 JSON 对象{}作为请求体；
   • 若显式设置Content-Type: application/x-www-form-urlencoded，运行时走表单分支而非默认 JSON 分支。
3. 对于POST+keep_params_as_form = true：
   • params合并进表单体；
   • 未显式设置 Content-Type 时自动加application/x-www-form-urlencoded；
   • body与params含相同 key 时，params的值覆盖body的值。
4. keep_page_param_as_http_param = true将分页字段直接写入params。
5. keep_page_param_as_http_param = false只更新 headers、params、body 中已有的 key 或占位符，不会自动新增分页字段。
6. pageing.use_placeholder_replacement = true支持${page}、${cursor}占位符，也支持带前后缀的替换（如"10${page}"在 page=5 时变成"105"）；为false时仅做基于 key 的替换。

上述规则的源码依据在 HttpClientProvider.execute() 与 HttpSourceReader.updateRequestParam()，其中第 2 条「默认非表单路径下 body 原样发送」由execute()开头的 verbatim 分支保证（避免把嵌套 JSON key 拍平）。

示例 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（默认非表单路径），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 表单提交，分页字段合并进表单体

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_field

当接口返回的是嵌套 JSON 结构时，可用content_field或json_field精准抽取目标片段，二者都以 Jayway JsonPath 实现（JsonPathProcessorImpl.java）。

content_field：整体提取 JSON 片段

content_field直接抽取一段 JSON 数据，无需为每个字段单独配置路径。例如只需store.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 } } }

对应的可运行测试配置见 http_contentjson_to_assert.conf，Mock 数据见 mockserver-config.json。

json_field：逐字段 JSONPath 映射

json_field帮助你把 schema 中每个字段映射到 JSONPath，必须与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的区别：前者按「字段 → JSONPath」逐个抽取并重组为行（数组模式下多条路径按索引对齐成多行），后者按单一 JSONPath 直接截取整块 JSON 再交给 schema 解析。二者的优先级处理在 HttpSourceReader.collect() 中：content_field优先于json_field。

当 JSONPath 匹配到的字段缺失时，默认会报错；设置json_filed_missed_return_null = true后缺失字段返回 null。该行为由 JsonPathProcessorImpl.java 控制，并有 JsonFieldMissedReturnNullTest.java 等测试覆盖。

七、分页拉取：pageing 完整指南

pageing目前支持两种分页类型：PageNumber（页码分页，默认）与Cursor（游标分页）。分页状态与终止判断的完整实现见 HttpSourceReader.internalPollNext()。

PageNumber 页码分页

使用PageNumber时，页码参数可以放在 HTTP 请求的不同位置：URL 参数（params）、请求体（bodyJSON）、请求头（headers）。配合use_placeholder_replacement = true，可用${page}占位符动态更新这些值，支持多种形态：

• 独立占位符："${page}"
• 带前后缀："10${page}"或"page-${page}"
• body 中不带引号的数字：{"a":${page},"limit":10}
• 嵌套 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={ #you can not set this parameter ,the default value is PageNumber page_type="PageNumber" total_page_size=20 page_field=page use_placeholder_replacement=true #when don't know the total_page_size use batch_size if read size<batch_size finish ,otherwise continue #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：基于 key 的替换（不使用占位符）

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

use_placeholder_replacement = false时走基于 key 的替换：params 中page的值被直接改写为当前页码（见 HttpSourceReader.processPageMap() 的map.containsKey(pageField)分支），body 则通过递归查找同名 key 改写（processBodyMapRecursively）。

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

source { Http { url = "http://localhost:8080/mock/queryData" method = "GET" format = "json" headers = { Page-Number = "10${page}" # Will become "105" when page=5 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 } } } }

带前后缀替换的实现是 HttpSourceReader.replacePlaceholder()：定位${page}占位符位置后，拼接前缀 + 页码 + 后缀。

示例 5：body 中不带引号的页码数字

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

示例 6：body 中嵌套 JSON 结构携带页码

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

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()：

• Cursor 模式：从响应中按cursor_response_field读取新游标；若游标为空或与上一轮相同，说明无法继续推进，结束拉取（noMoreElementFlag = true）。
• PageNumber + total_page_size > 0：当前页号 >= total_page_size时结束。
• PageNumber + total_page_size = 0：统计当前页实际返回的行数（对解析后的 JSON 数组size()），当读取行数 < batch_size时结束，等于batch_size时继续下一页。

页码推进循环在internalPollNext()中：PageNumber每轮pageIndex += 1后重新调用updateRequestParam更新请求；两轮之间Thread.sleep(10)做轻微限速。仓库 e2e 用例 http_page_increase_page_num.conf 演示了固定total_page_size = 2的页码递增场景，并断言最终恰好输出 4 行；http_page_cursor_num_assert.conf 覆盖游标分页场景。

八、可靠性保障：超时与重试

连接超时与 Socket 超时

• connect_timeout_ms：建立 TCP 连接的超时时间，默认 12000ms（12 秒）；
• socket_timeout_ms：等待响应的 Socket 超时，默认 60000ms（60 秒）。

二者在 HttpClientProvider 构造时写入RequestConfig，作用于该连接器发起的每个请求。

重试策略

• retry：最大重试次数。注意：仅当请求抛出IOException时才触发重试，非 IOException（如业务错误码）不重试。源码中retryIfException(ex -> ExceptionUtils.indexOfType(ex, IOException.class) != -1)。
• retry_backoff_multiplier_ms：退避基准，默认 100ms；
• retry_backoff_max_ms：退避上限，默认 10000ms。

重试等待采用斐波那契退避（WaitStrategies.fibonacciWait(multiplier, max, MILLISECONDS)），配合StopStrategies.stopAfterAttempt(retry)控制总尝试次数，失败时通过RetryListener打印告警日志。若retry < 1，则构建一个不重试的 Retryer。

配置示例：

Http { url = "https://api.example.com/data" method = "GET" retry = 3 retry_backoff_multiplier_ms = 200 retry_backoff_max_ms = 5000 }

九、流式轮询：poll_interval_millis

在job.mode = "STREAMING"下，Http Source 每完成一轮拉取后若没有signalNoMoreElement，会Thread.sleep(pollIntervalMillis)后继续请求（见 HttpSourceReader.internalPollNext() 的 finally 分支）。配合enable_multi_lines可逐行处理持续追加的文本流；配合 Kafka、JDBC 等 sink 即可构建实时数据管道。

一个典型的流式配置骨架：

env { parallelism = 1 job.mode = "STREAMING" } source { Http { url = "http://example.com/api/events" method = "GET" format = "text" enable_multi_lines = true poll_interval_millis = 5000 } } sink { Console {} }

十、e2e 测试与排障建议

仓库在 connector-http-e2e 提供了完整的端到端测试，覆盖本连接器几乎所有核心场景，可作为配置正确性的参考实现：

• http_json_to_assert.conf：JSON 格式 + schema 解析；
• http_contentjson_to_assert.conf：content_field提取；
• http_jsonpath_to_assert.conf：json_field提取；
• http_binary_to_assert.conf：binary 文件下载；
• http_page_increase_page_num.conf：页码分页；
• http_page_cursor_num_assert.conf：游标分页；
• http_multilinejson_to_assert.conf：多行 JSON。

这些用例的 Mock 服务端数据统一由 mockserver-config.json 提供，connector-http-e2e的测试类（如HttpIT）会启动 MockServer 后提交作业并断言输出。

排障自查清单：

1. 请求形态与预期不符：按第五节「分页与最终请求形态速查」从 url、headers、body 三个维度反推最终发出的请求，重点检查keep_params_as_form与keep_page_param_as_http_param两个兼容开关。
2. 分页不结束或提前结束：确认total_page_size是否设置；未设置时确认返回的 JSON 数组长度是否满足batch_size判断逻辑（读取行数 < batch_size才结束）。
3. JSON 解析报错：format = json必须配置schema；字段缺失时按需开启json_filed_missed_return_null = true。
4. binary 模式报错：确认作业为 BATCH 模式、未配置pageing、schema 固定为三列(data, relativePath, partIndex)。
5. 接口慢导致超时：调大connect_timeout_ms/socket_timeout_ms，或配置retry相关参数。
6. pageing拼写：保持pageing（双 e）拼写，这是连接器选项名的既定设计。

通过本文的配置全解与源码级原理解析，你可以快速上手 Http Source 连接器，并在分页、JSON 提取、二进制下载等复杂场景下准确配置、高效排障。

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