数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载导读本文围绕 mage-ai 开源仓库中 API 数据源文档 展开系统讲解如何在 Mage 数据集成Data Integration管道中接入任意 HTTP REST API——包括 GET / POST 请求、嵌套 JSON 字段提取、CSV / TSV / XLSX 文件读取以及响应解析与列定义规则。读完本文你将掌握 API 数据源全部 9 个配置项的含义与默认值、4 类典型场景的完整可运行配置并能理解底层源码的解析逻辑从而把任意第三方 API 或内部服务接口稳定地同步到目标数据仓库。一、API 数据源在 Mage 中的定位在 mage-ai 的数据集成体系中API是一个通用数据源Source它把「发起 HTTP 请求并解析响应」这一能力抽象成标准的 Source 接口因此任何一个返回 JSON、CSV、TSV 或 XLSX 的 HTTP 接口都可以通过它接入数据管道。其核心实现在 mage_integrations/mage_integrations/sources/api/init.py 中对应的 Source 类名为Api继承自 mage_integrations/mage_integrations/sources/base.py 中的通用Source基类。从源码结构看该数据源具有以下特征输出流stream固定命名为api复制方式固定为FULL_TABLE全量不维护增量水位线key_properties 为空唯一冲突处理方式为UPDATE支持discover自动推断表结构、test_connection连通性检查、load_data批量取数三个阶段完整符合 Mage 数据集成 Source 的标准生命周期。二、配置参数全解配置该数据源时需要填写以下凭据与行为参数下表为原文档的完整参数清单并结合源码补充了默认值与行为细节Key说明是否必填urlAPI 地址完整 URL含路径与必要的既有 query必填methodGET或POST可选默认GETqueryURL 查询参数会与 URL 中已有的 query 合并可选payload当 method 为POST时作为请求体发送的数据可选headers请求头会与内置的user-agent合并可选response_parser使用点号dot notation解析 API 响应最终结果必须是数组可选columns当 API 或response_parser返回的最终数据不是 JSON 对象如字符串数组、数组的数组时必须定义列名条件必填separator当响应为 TSV、XLSX 或使用了特殊分隔符的 CSV 时指定默认,可选has_header当响应为 TSV、XLSX 或 CSV 且含表头时设为True默认False可选源码层面的补充说明method 归一化http_method属性按self.config.get(method, GET).upper() POST判断即大小写不敏感缺省视为 GET见init.py 中http_method定义。query 合并逻辑配置中的query字典会通过parse_qs解析 URL 中已存在的查询参数并做合并再经urlencode重新拼接到 URL 上避免覆盖原有参数。headers 合并逻辑默认内置一个浏览器风格的user-agent配置的headers在其之上合并覆盖。columns 兜底当数据不是 JSON 对象且未配置columns时源码会自动生成col0、col1… 这样的列名兜底细节见后文「源码级实现原理」。separator与has_header的默认值分别由load_data中的separator ,与header False兜底与表格描述一致。对应的配置模板位于 mage_integrations/mage_integrations/sources/api/templates/config.json展示了所有字段的默认形态{ columns: null, headers: null, method: GET, payload: null, query: null, response_parser: null, url: }三、示例一GET response_parser 提取嵌套字段当 API 返回的是多层嵌套 JSON而真正需要同步的数据深藏在某个嵌套节点中时response_parser是提取它的关键。以下示例配置原文完整保留{ url: https://api.plos.org/search, query: { q: title:DNA }, headers: { Content-Type: application/json }, response_parser: response.docs[0].author_display, columns: [author_display] }该接口返回的响应是一个典型的嵌套 JSON{ response: { numFound: 5669, start: 0, maxScore: 6.7217336, docs: [ { id: 10.1371/journal.pone.0000290, journal: PLoS ONE, eissn: 1932-6203, publication_date: 2007-03-14T00:00:00Z, article_type: Research Article, author_display: [ Rayna I. Kraeva, Dragomir B. Krastev, Assen Roguev, Anna Ivanova, Marina N. Nedelcheva-Veleva, Stoyno S. Stoynov ], abstract: [ Nucleic acids, due to their structural and chemical properties, can form double-stranded secondary structures that assist the transfer of genetic information and can modulate gene expression. However, the nucleotide sequence alone is insufficient in explaining phenomena like intron-exon recognition during RNA processing. This raises the question whether nucleic acids are endowed with other attributes that can contribute to their biological functions. In this work, we present a calculation of thermodynamic stability of DNA/DNA and mRNA/DNA duplexes across the genomes of four species in the genus Saccharomyces by nearest-neighbor method. The results show that coding regions are more thermodynamically stable than introns, 3′-untranslated regions and intergenic sequences. Furthermore, open reading frames have more stable sense mRNA/DNA duplexes than the potential antisense duplexes, a property that can aid gene discovery. The lower stability of the DNA/DNA and mRNA/DNA duplexes of 3′-untranslated regions and the higher stability of genes correlates with increased mRNA level. These results suggest that the thermodynamic stability of DNA/DNA and mRNA/DNA duplexes affects mRNA transcription. ], title_display: Stability of mRNA/DNA and DNA/DNA Duplexes Affects mRNA Transcription, score: 6.7217336 } ] } }然而由于配置了response_parser为response.docs[0].author_display真正从响应中提取出的数据是[ Rayna I. Kraeva, Dragomir B. Krastev, Assen Roguev, Anna Ivanova, Marina N. Nedelcheva-Veleva, Stoyno S. Stoynov ]注意response_parser的解析规则支持点号路径 数组下标组合。上面的response.docs[0].author_display含义为先取响应中的response对象再取其中的docs数组取下标为0的元素再取该元素的author_display字段。其底层实现位于 mage_integrations/mage_integrations/utils/dictionary.py 的dig函数它会把字符串按.拆分为逐级路径并用正则\[(\d)\]$识别每一级末尾的数组下标。由于最终数据中的每一项如Rayna I. Kraeva是字符串而非 JSON 对象此时必须提供columns配置。最终数据在输出到目标端之前会被转换为 JSON 对象数组[ { author_display: Rayna I. Kraeva }, { author_display: Dragomir B. Krastev }, { author_display: Assen Roguev }, { author_display: Anna Ivanova }, { author_display: Marina N. Nedelcheva-Veleva }, { author_display: Stoyno S. Stoynov } ]四、示例二GET 直接返回 JSON 数组当接口本身直接返回 JSON 对象数组时不需要response_parser也不需要columns。以下配置原文完整保留{ url: https://api.coingecko.com/api/v3/coins/markets, query: { vs_currency: usd } }接口返回的响应为[ { id: bitcoin, symbol: btc, name: Bitcoin, image: https://assets.coingecko.com/coins/images/1/large/bitcoin.png?1547033579, current_price: 17154.23, market_cap: 329331527834, market_cap_rank: 1, fully_diluted_valuation: 359638803581, total_volume: 14339246353, high_24h: 17227.56, low_24h: 17107.75, price_change_24h: 18.57, price_change_percentage_24h: 0.10839, market_cap_change_24h: -425260465.15130615, market_cap_change_percentage_24h: -0.12896, circulating_supply: 19230300.0, total_supply: 21000000.0, max_supply: 21000000.0, ath: 69045, ath_change_percentage: -75.16662, ath_date: 2021-11-10T14:24:11.849Z, atl: 67.81, atl_change_percentage: 25185.95387, atl_date: 2013-07-06T00:00:00.000Z, roi: null, last_updated: 2022-12-11T00:32:01.695Z } ]因为这里没有配置response_parser最终数据与 API 的响应完全一致。同时由于数组中的每一项本身就是 JSON 对象columns配置项就不再需要了。五、示例三POST 请求携带请求体对于需要 POST 提交数据的接口通过method: POST与payload发送请求体原文完整保留YAML 格式便于在界面配置中阅读url: https://api.something.com/users method: POST payload: user: first_name: Urza power: 10 headers: Content-Type: application/json token: abc123 response_parser: user该示例同时展示了三点组合用法method: POST触发http_method返回post请求以 POST 发出payload中的结构化数据作为请求体源码中通过datapayload传给requestsheaders中的Content-Type: application/json与token等鉴权信息会与默认user-agent合并response_parser: user从返回的 JSON 中取出user节点其结果必须是数组才能作为记录流输出。六、示例四直接读取 CSV / TSV / XLSXGoogle Sheets 导出API 数据源不仅能处理 JSON还能直接消费以 HTTP 形式暴露的表格文件。最典型的场景是把 Google Sheets 以outputcsv/outputtsv/outputxlsx方式导出配置示例如下原文完整保留url: https://docs.google.com/spreadsheets/d/e/2PACX-1vTLcLUBAJAWf-8NQSjlbB3E4LR6DWk5QIZC-KtRb1j2CXXcgY6mE6vOJAW8PoJ1BAOgjXYpE4tY1LAD/pub?outputxlsx method: GET has_header: True该接口返回的是带表头的电子表格内容示例中为 CSV 文本形态teste,first_name,second_name,email 1,Sadella,Tythacott,stythacott0sina.com.cn 2,Melessa,Flaune,mflaune1si.edu 3,Caroljean,Filipowicz,cfilipowicz2guardian.co.uk 4,Doll,Wannan,dwannan3people.com.cn 5,Nancy,Giraudy,ngiraudy4pagesperso-orange.fr 6,Dominic,Bimson,dbimson5vinaora.com 7,Kikelia,Bishopp,kbishopp6cdc.gov 8,Andrus,Pomfrett,apomfrett7wikipedia.org 9,Wildon,Fillingham,wfillingham8google.co.uk 10,Alfonse,Leechman,aleechman9jalbum.net 11,Phil,Emblem,pemblemaopera.com 12,Eyde,Brewer,ebrewerbistockphoto.com ....此时has_header: True表示第一行是列名separator缺省为,即 CSV 逗号分隔。若为 TSV则需设置separator: \t。在界面上配置完成后还可以点击Load Sample data按钮直接预览最终抽取出来的数据行用于校验配置是否正确。七、源码级实现原理7.1 请求构建__build_responseApi类的__build_response方法完成请求的组装与发送关键行为包括参数合并将query与 URL 中已有查询参数合并后重新拼接将配置headers合并进内置的user-agent默认头重试策略通过requests.Session()分别对http://与https://挂载max_retries100的HTTPAdapter具备极强的网络容错能力超时与校验请求timeout12且verifyFalse跳过 TLS 证书校验适合内网自签名证书场景生产环境需注意安全性取舍连通性测试test_connection直接复用该方法发起请求并断言status_code 200否则抛出异常。7.2 响应类型识别与分发_check_response_type/load_dataload_data是取数核心它先通过_check_response_type判定响应类型再进行分发Google Sheets URL以https://docs.google开头时走专门分支按outputcsv/outputtsv/outputxlsx分别用polars.read_csv、polars.read_csv(separator\t)、pd.read_excel解析text/plain、text/csv用polars.read_csv支持separator、has_header解析后转为 recordsapplication/gzip解压后按 CSV 读取application/json先response.json()若配置了response_parser则调用dig提取随后判断数组首元素是否为 dict 来决定是否触发columns逻辑其他 MIME 类型尝试pd.read_excel失败则提示确认扩展名是否为 XLSX。7.3columns的条件触发与自动兜底当 JSON 数据首元素不是 dict例如字符串数组或嵌套数组的数组时requires_columns True。此时若配置了columns按item[idx]逐列取值填充不足的列补None若未配置columns源码会自动按最长子数组的长度生成col0、col1… 作为兜底列名——这是界面上「columns 条件必填」背后的实际行为。7.4 结构发现discover与目录生成discover方法会实际调用load_data拉取样本行用pandas.DataFrame结合infer_dtypes推断每列类型遇到mixed混合类型时按出现频率最高的类型归类list→ ARRAY、dict→ OBJECT、其余 → STRING最终生成 stream id 为api、复制方式为FULL_TABLE、无主键、冲突处理为UPDATE的标准 Catalog供目标端建表使用。这一推断链路在 test_api.py 中有完整的断言佐证。八、测试用例与验证依据仓库在 mage_integrations/mage_integrations/tests/sources/api/test_api.py 中提供了该数据源的单元测试覆盖四种典型场景test_api_csvGoogle Sheetsoutputcsvhas_headerTrue断言 discover 得到的 Catalog 中teste为null/integer、first_name/second_name/email为null/stringtest_api_tsvoutputtsv场景Catalog 与 CSV 场景一致test_api_xlsxoutputxlsx场景test_api_jsonGET JSON 接口断言 24 个字段的类型推断字符串、number、integer、object 等与market_cap_rank为null/integer等细节。这些测试既是对该数据源行为的事实性约束也直接说明了「接口返回什么形态、配置如何写、最终 Catalog 长什么样」三者间的对应关系可作为自行接入新 API 时的对照模板。数据源以独立可执行入口方式运行if __name__ __main__: main(Api)也支持通过discover、test_connection、load_sample_data等标准模式被 Mage 平台调用。小结接入一个 API 数据源的完整决策路径可以总结为四步① 明确响应格式JSON 数组 / 嵌套 JSON / CSV / TSV / XLSX→② 决定是否需要response_parser嵌套提取才需要且最终结果必须是数组→③ 判断是否需要columns最终数据每项不是 JSON 对象才需要→④ 按需设置separator与has_header表格文件场景。结合 API 数据源文档、核心实现、配置模板 与 测试用例即可在 Mage 中快速、稳定地把任意 HTTP 数据源接入数据管道。赞分享数据工程数据编排ETL任务调度批处理流处理数据集成后端【免费下载链接】mage-ai Build, run, and manage data pipelines for integrating and transforming data.项目地址https://gitcode.com/gh_mirrors/ma/mage-ai点击查看免费下载相关推荐使用 Mage-ai 将 Salesforce 数据接入数据管道Source 配置、OAuth 认证与 REST/BULK API 同步实战使用 Mage ai 将 Salesforce 数据接入数据管道Source 配置、OAuth 认证与 REST/BULK API 同步实战 本篇技术指南聚焦数据工程数据编排ETL任务调度批处理流处理数据集成后端前端使用 dlt REST API Source 构建数据管道从 REST 接口加载数据到 DuckDB 实战教程使用 dlt REST API Source 构建数据管道从 REST 接口加载数据到 DuckDB 实战教程 本教程围绕 dlt 的 REST API So数据工程数据集成批处理SeaTunnel Firebase Source Connector 实战通过 REST API 将 Firebase Realtime Database 数据接入数据管道SeaTunnel Firebase Source Connector 实战通过 REST API 将 Firebase Realtime Database数据集成ETL大数据批处理流处理变更数据捕获上一篇FastStream项目自定义AsyncAPI文档完全指南下一篇如何在15分钟内使用Liveblocks与Next.js构建实时协作应用完整指南 创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考