SeaTunnel Github Source Connector 实战指南基于 HTTP 连接器对接 GitHub REST API【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文聚焦 Apache SeaTunnel 的GithubSource 连接器位于 connector-http-github 模块讲解如何通过它从 GitHub REST API 批量或流式读取数据。读完本文你将掌握Github连接器的全部配置参数、两种分页模式页码分页与游标分页的用法、json格式下 schema 的定义方式以及三个可直接复用的任务配置示例并能结合源码理解其底层实现原理。连接器概述GithubSource 连接器用于从 GitHub REST API 读取数据如组织仓库列表、仓库事件流等。它本质上是 HTTP Source 连接器的一个特化实现源码中GithubSource extends HttpSource见 GithubSource.java插件名PLUGIN_NAME Github因此在任务配置中 source 类型写为Github。与通用 HTTP 连接器相比它唯一的增强点在于自动为每个请求添加Authorization: Bearer access_token请求头免去手动拼接鉴权头的麻烦。核心能力与特性矩阵Github连接器继承自 HTTP 连接器的能力支持特性如下详见 connector-v2-features特性支持情况批处理batch✅ 支持流处理stream✅ 支持通过poll_interval_millis轮询精确一次exactly-once❌ 不支持列投影column projection❌ 不支持并行度parallelism❌ 不支持用户自定义分片user-defined split❌ 不支持需要说明的是该连接器目前为单分片single split读取模式并行度默认即 1任务配置中env.parallelism通常保持为 1。工作原理access_token 如何变成 Bearer 头这是Github连接器区别于通用 HTTP 连接器的核心机制其实现链路非常清晰任务配置解析阶段GithubSourceFactoryGithubSourceFactory.java通过getHttpBuilder().required(GithubSourceOptions.ACCESS_TOKEN)将access_token声明为必填项未配置会直接报错。参数构建阶段GithubSourceParameter extends HttpParameterGithubSourceParameter.java在buildWithConfig中检测到access_token后调用formatOauthToken将其格式化为Bearer accessToken写入headers的Authorization键。最终所有 HTTP 请求都会携带Authorization: Bearer token头符合 GitHub REST API 的 OAuth2 鉴权要求。相关的常量定义在 GithubSourceOptions.javaAUTHORIZATION_KEY Authorization、BEARER_KEY Bearer。该类的单元测试见 GithubFactoryTest.java。Source 配置参数详解下表列出了GithubSource 的全部可配置参数其中大部分继承自 HTTP 连接器底层定义见 HttpSourceOptions.java 与 HttpCommonOptions.java参数名类型必填默认值urlString是-access_tokenString是-methodString否GETheadersMap否-paramsMap否-bodyString否-formatString否textschemaConfig否-schema.fieldsConfig否-json_fieldConfig否-content_fieldString否-pageingConfig否-page_typeString否PageNumbercursor_fieldString否-cursor_response_fieldString否-poll_interval_millisint否-retryint否-retry_backoff_multiplier_msint否100retry_backoff_max_msint否10000enable_multi_linesboolean否falsekeep_params_as_formboolean否falsekeep_page_param_as_http_paramboolean否falsebatch_sizeint否100start_page_numberlong否1total_page_sizelong否0use_placeholder_replacementboolean否falseconnect_timeout_msint否12000socket_timeout_msint否60000json_filed_missed_return_nullboolean否falsecommon-optionsconfig否-请求相关参数url [String]必填GitHub REST API 地址例如https://api.github.com/orgs/apache/repos。access_token [String]必填GitHub 个人访问令牌Personal Access Token。连接器会将其以 Bearer token 形式放入 HTTPAuthorization请求头。GitHub 对未鉴权的 API 请求有严格的速率限制生产环境务必配置有效的 token。method [String]HTTP 请求方法GitHub 常见读取场景使用GET底层HttpRequestMethod枚举默认GET。headers [Map]额外自定义请求头。注意不要在headers中手动配置Authorization除非你有意覆盖由access_token自动生成的鉴权头。params [Map]HTTP 查询参数例如per_page、page、since等 GitHub API 参数。body [String]HTTP 请求体仅对接受请求体的 API 端点有用。connect_timeout_ms [int]HTTP 连接超时时间毫秒默认 12000源码常量DEFAULT_CONNECT_TIMEOUT_MS 6000 * 2。socket_timeout_ms [int]HTTP Socket 超时时间毫秒默认 60000源码常量DEFAULT_SOCKET_TIMEOUT_MS 6000 * 10。响应解析相关参数format [String]响应格式支持json与text默认text。需要输出带字段名的 SeaTunnel 行时使用format json并配合schema。schema [Config]当format json时定义输出行结构完整说明见 schema-feature。json_field [Config]将输出字段映射到 JSONPath 表达式当所需值嵌套在响应中时与schema配合使用。content_field [String]在 schema 解析前先通过 JSONPath 选取 JSON 片段例如$.items[*]GitHub 返回的列表类响应常用此写法。enable_multi_lines [boolean]为true时响应体中用换行分隔的多个 JSON 对象会被视为多条独立记录。json_filed_missed_return_null [boolean]为true时缺失的 JSON 字段返回null否则缺失字段会报错。分页相关参数pageing [Config]从 HTTP 连接器继承的分页配置块任务配置中必须保留pageing这个拼写。需要基于游标的 GitHub 分页时在其中配置page_type Cursor。page_type [String]分页类型支持PageNumber默认与Cursor。返回next游标的端点如 GitHub Events API应使用Cursor。cursor_field [String]携带游标值的请求参数名与page_type Cursor配合使用。cursor_response_field [String]响应体中游标的 JSONPath与page_type Cursor配合使用。batch_size [int]当总页数未知时每个分页请求返回的记录数默认 100。start_page_number [long]从第几页开始同步默认 1。total_page_size [long]读取的总页数0表示使用batch_size一直读取到 API 不再返回新页为止。use_placeholder_replacement [boolean]为true时使用${field}占位符替换 headers、参数和 body 值否则使用基于 key 的替换方式。在分页场景中用于把页码注入请求参数。keep_params_as_form [boolean]为true时请求参数以表单编码形式放入 body 而非 URL 查询参数。keep_page_param_as_http_param [boolean]为true时分页时页码参数保留在请求 URL 中而不是被替换进 body。流式与重试参数poll_interval_millis [int]流式任务中两次请求的间隔毫秒。批任务只读一次即结束无需配置。retry [int]HTTP 请求抛出IOException时的最大重试次数。retry_backoff_multiplier_ms [int]重试退避倍数毫秒默认 100。retry_backoff_max_ms [int]最大重试退避时间毫秒默认 10000。common optionsSource 插件通用参数见 source-common-options。分页机制深入页码分页PageNumber传统页码分页是默认模式page_type默认值为PageNumber见 HttpSourceOptions.java。核心要点页码参数名由page_field指定默认值为page起始页码由start_page_number控制默认 1总页数由total_page_size控制0表示未知、按batch_size持续读取在params中通过page ${page}占位符注入页码同时需开启use_placeholder_replacement true。游标分页Cursor针对 GitHub 返回next游标的端点如 Events API配置page_type Cursor、cursor_field请求参数名与cursor_response_field响应中游标的 JSONPath。底层实现位于 HttpSourceReader.java每次请求后从响应中按cursor_response_field提取新游标若新游标为空或与当前游标相同Objects.equals(currentCursor, newCursor)则判定分页结束。任务示例以下示例均来自官方文档并保持完整可用sink 统一使用 Console 便于快速验证输出。示例一读取 GitHub 组织的仓库列表批处理env { parallelism 1 job.mode BATCH } source { Github { url https://api.github.com/orgs/apache/repos access_token ghp_xxxxxxxxxxxx method GET format json schema { fields { id int name string description string html_url string stargazers_count int forks int } } } } sink { Console { } }要点format json配合schema定义输出字段GitHub 组织仓库列表接口直接返回 JSON 数组无需content_field。示例二读取分页的 GitHub API 结果env { parallelism 1 job.mode BATCH } source { Github { url https://api.github.com/orgs/apache/repos access_token ghp_xxxxxxxxxxxx method GET params { per_page 100 page ${page} } pageing { page_field page total_page_size 5 start_page_number 1 use_placeholder_replacement true } format json schema { fields { id int name string html_url string } } } }要点params中的page ${page}占位符由分页逻辑逐页替换total_page_size 5表示最多拉取 5 页per_page 100每页取 100 条即本任务最多读取 500 条记录。示例三流式读取 GitHub 组织事件Streamingenv { parallelism 1 job.mode STREAMING checkpoint.interval 30000 } source { Github { url https://api.github.com/orgs/apache/events access_token ghp_xxxxxxxxxxxx method GET format json poll_interval_millis 60000 schema { fields { id string type string created_at string } } } }要点job.mode STREAMING开启流模式poll_interval_millis 60000表示每 60 秒轮询一次 GitHub Events API建议配合checkpoint.interval做状态恢复。若需游标分页读取历史事件可将pageing.page_type设为Cursor并配置cursor_field/cursor_response_field。使用建议与注意事项token 安全管理access_token属于敏感信息应避免在共享任务文件中硬编码真实 token推荐使用 SeaTunnel 的变量替换机制或部署平台的密钥管理能力。鉴权头来源连接器总是由access_token生成Authorization: Bearer access_token头其他自定义头一律放入headers。类型化输出需要输出带类型的 SeaTunnel 行时设置format json并定义schema。嵌套数组处理当 GitHub 响应把记录包在嵌套数组中时如$.items[*]使用content_field提取。分页模式选择传统页码分页保持page_type PageNumber并在params中使用page/per_page游标类端点如 GitHub Events API则设置page_type Cursor并配置cursor_field/cursor_response_field。速率限制GitHub REST API 存在速率限制流式任务请合理设置poll_interval_millis并利用retry、retry_backoff_multiplier_ms、retry_backoff_max_ms应对瞬时故障。相关资源连接器模块源码connector-http-github底层 HTTP 连接器参数定义HttpSourceOptions.java分页与读取核心实现HttpSourceReader.java连接器通用能力说明connector-v2-featuresSchema 定义完整指南schema-feature【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考