在实际机器学习项目中数据获取很少是从现成 CSV 文件开始的。更多情况是目标数据在业务系统、第三方平台或公开数据服务中需要先通过 API 按条件取回来再进入清洗和建模流程。在 100 天机器学习计划的第 17 天主题就是“从 API 获取数据”。这一天的练习不是去背某个接口文档而是建立一套可以复用的能力看懂响应结构、正确构造请求、处理分页和失败、把 JSON 转成 DataFrame、再安全地落到磁盘。接下来会从最小请求开始一步步写到一个可复用的获取函数并给出常见报错的排查清单。学完之后你可以把同样的方法迁移到自己的项目里只要把 URL、参数和字段名换成实际业务接口即可。1. 先理清 API 在机器学习数据链路中的位置1.1 机器学习项目的数据来源有哪些机器学习项目的数据来源并不单一。常见来源包括公开数据集平台、关系型数据库、日志文件、图像目录、爬虫抓取结果以及各类 HTTP API 服务。不同来源在获取方式、时效性、权限控制和使用成本上差异很大。数据来源获取方式优点缺点典型场景公开数据集直接下载 CSV、Parquet简单、标准化更新慢、字段固定入门练习、Kaggle 比赛关系型数据库SQL 查询、ETL数据质量可控需要连接权限和结构了解内部业务数据建模日志文件文件采集、解析覆盖全、历史多格式杂、噪音大用户行为分析爬虫抓取解析 HTML 或模拟请求可定制强受站点结构影响、维护成本高补充公开网页信息HTTP API按接口请求 JSON实时、带权限、适合系统集成受限流和频率限制业务系统、天气、行情、地图等在机器学习流程里使用 API 获取数据通常不是为了替代其它方式而是为了拿到实时或准实时的结构化数据。比如天气预测需要每日温度、湿度推荐系统需要用户最新行为事件风险模型需要按用户身份查询关联信息。这些数据如果只依赖离线导出等数据到位时可能已经失去时效性。1.2 API 和爬虫、数据库导出的边界在哪里API 是服务方公开的、有明确请求格式的数据接口。它和爬虫的主要区别在于API 是官方提供的数据通道有文档约束请求参数和返回结构爬虫需要自己解析页面结构且页面一旦改版解析逻辑通常要跟着改。API 也更适合做系统间集成因为响应通常是 JSON 或 XML可以直接进入下游处理。但不要把 API 看作万能方案。如果是静态大文件比如几 GB 的历史语料或图像压缩包直接用对象存储链接批量下载更合适如果数据已经存在于数据仓库中使用 SQL 抽取会比逐条请求 API 快得多。API 更适合按需查询、实时同步、以及有权限控制的增量数据获取。1.3 什么样的数据适合通过 API 获取适合通过 API 获取的数据有几个共同特征数据总量可能不大但请求条件灵活数据需要频繁更新数据存放位置在远端系统无法直接连接数据库数据需要附带身份权限只允许授权应用访问。一个典型的例子是天气数据。你可以通过经纬度、日期范围和字段列表请求某一段时间内的温度、降水量、风速。返回的 JSON 结构稳定字段符合业务语义保存成 CSV 后可以直接用于时间序列预测练习。这类数据非常适合作为第 17 天的实践对象因为不需要额外申请复杂权限响应结构也容易理解。2. 配置 Python 环境先把依赖和目录准备好2.1 确认 Python 版本并安装依赖从 API 获取数据并不需要复杂框架Python 配合requests和pandas就能完成大部分工作。先确认当前 Python 版本再安装依赖。python --version pip install requests pandas pyarrowrequests负责发送 HTTP 请求和处理响应pandas负责把 JSON 转为 DataFrame并完成后续落盘pyarrow是可选依赖只有需要保存为 Parquet 格式时才需要安装。下面是一组参考版本不是硬性要求落地前以你所在环境实际可用的版本为准。组件参考版本说明Python3.10 或 3.113.8 以下版本不建议新项目使用requests2.31.x构造 HTTP 请求和处理状态码pandas2.1.xDataFrame 处理、CSV 与 Parquet 读写pyarrow14.x 或更高保存 Parquet 时的底层格式支持2.2 推荐目录结构原始数据、处理后数据、日志分开在写取数脚本之前先约定目录结构可以避免数据污染。常见做法是把原始响应、处理后数据、脚本和日志分开。ml-api-data/ ├── data/ │ ├── raw/ # 接口原始响应尽量不修改 │ └── processed/ # 清洗、对齐字段后的数据 ├── scripts/ │ └── fetch_data.py ├── logs/ │ └── fetch.log └── .env.exampleraw目录保存从 API 拿到的原始数据processed目录保存后续清洗、转换后的版本。这样当清洗逻辑出错时可以随时从原始数据重新处理不需要重新调用接口。logs目录保存运行日志便于在定时任务失败时定位问题。.env.example用来登记环境变量项比如 API Token 的键名方便团队成员了解需要配置哪些内容。2.3 先做一次连通性验证再开始写取数逻辑正式写函数之前先用最短代码验证请求链路是否可用。这样可以区分网络问题、参数问题、代码问题。import requests url https://jsonplaceholder.typicode.com/posts resp requests.get(url, timeout10) print(resp.status_code) print(resp.text[:300])这段代码做了三件事请求指定接口、设置 10 秒超时、打印状态码和响应前 300 个字符。如果状态码不是 200后续写再多逻辑也无法正常推进。这里选择的示例接口是公开的占位 API专门用于教学测试。如果你在特定网络环境下访问不了该域名可以换成自己项目中可访问的接口地址。注意不要省略timeout。不设置超时会导致请求长时间阻塞脚本一旦卡住排错难度会明显上升。3. 写一个最小案例从公开 API 拉取数据并转成 DataFrame3.1 先看一个公开 API 的响应结构以 JSONPlaceholder 的/posts接口为例请求成功后返回一个 JSON 数组每个元素是一个文章对象包含userId、id、title、body四个字段。这个结构很适合展示从 JSON 到 DataFrame 的转换流程。{ userId: 1, id: 1, title: sunt aut facere repellat provident occaecati excepturi optio reprehenderit, body: quia et suscipit\nsuscipit recusandae consequuntur expedita et cum\nreprehenderit molestiae ut ut quas totam\nnostrum rerum est autem sunt rem eveniet architecto }在写数据处理代码前先确认响应是 JSON 对象还是 JSON 数组。如果是数组转 DataFrame 时直接pd.DataFrame(data)如果是单个对象可能需要先包成列表或者按字段提取后再处理。拿到的数据结构决定了后续代码的写法这是最容易忽略的细节。3.2 发送 GET 请求并检查状态码使用requests.get发送请求并通过状态码判断请求是否成功。import requests import pandas as pd url https://jsonplaceholder.typicode.com/posts resp requests.get(url, timeout10) if resp.status_code ! 200: print(f请求失败状态码: {resp.status_code}) print(resp.text[:500]) else: data resp.json() df pd.DataFrame(data) print(shape:, df.shape) print(df.head(3))resp.json()会把响应内容解析为 Python 的 list 或 dict。对于 JSONPlaceholder 的/posts接口解析出的 list 中每个元素是一个 dict因此pd.DataFrame(data)能直接得到表格。预期输出如下shape: (100, 4) userId id title body 0 1 1 sunt aut facere... quia et suscipit... 1 1 2 qui est esse... est rerum tempore... 2 1 3 ea molestias quasi... et iusto sed quo...这里出现了一个判断习惯只有状态码为 200 时才继续解析。不要无条件调用resp.json()否则遇到错误响应时会因为 JSON 解析失败让报错信息失去定位价值。3.3 把 JSON 响应转换为 DataFrame 并完成初步校验得到 DataFrame 后先不要急着保存或建模至少完成三项检查行数是否合理、字段是否齐全、字段类型是否符合预期。print(总行数:, len(df)) print(字段列表:, list(df.columns)) print(df.dtypes)如果接口返回 100 条数据len(df)就应该是 100。字段列表应该包含文档中声明的字段。dtypes能帮你发现数据被解析成了字符串、整数还是对象类型。有些接口会把数字字段返回成字符串比如金额、身份证号、号码这种情况在后续特征工程中需要先做类型转换。4. 处理真实场景中更常见的 API 细节4.1 分页大结果集不能只取第一页一次请求能返回的数据量通常有限接口服务方都会设置分页参数。常见的分页方式有三种页码分页、游标分页和时间窗口分页。分页模式典型参数适合场景注意事项offset / pagepage2limit20数据总量可控新增或删除数据时可能重复或遗漏cursorcursorabc123高吞吐、数据持续变化不能随意跳页要看 response 中的 next 字段time rangestart_time/end_time日志、指标、时间序列注意时区、时间跨度上限、边界重叠仍然以 JSONPlaceholder 为例它支持_page和_limit参数。写一个循环可以抓取多页数据。import requests def fetch_posts_paged(base_url, page_size20, max_pages5): rows [] page 1 while True: params {_page: page, _limit: page_size} resp requests.get(base_url, paramsparams, timeout10) resp.raise_for_status() batch resp.json() rows.extend(batch) if len(batch) page_size: break if max_pages and page max_pages: break page 1 return rows这段逻辑的关键不是抓完所有页而是要有退出条件。如果接口在最后一页返回少于page_size的数据说明没有下一页如果设置了max_pages则必须做页数上限保护。否则遇到参数理解错误时很容易变成死循环。4.2 认证、参数、请求头与超时控制许多真实 API 不能匿名访问。常见认证方式是使用请求头Authorization: Bearer token。Token 通常需要从服务方控制台申请并注意有效期。headers { Authorization: Bearer your_token_here, Content-Type: application/json } resp requests.get(url, headersheaders, timeout10)另外要注意不要把 Token 直接硬编码在脚本里更不要提交到 Git 仓库。更好的做法是放到环境变量或.env文件中并且让.env文件进入忽略列表。示例代码如下import os token os.getenv(API_TOKEN) if not token: raise RuntimeError(缺少 API_TOKEN请在环境变量中配置)超时也应该更细粒度地设置。timeout(5, 15)表示连接超时 5 秒读取超时 15 秒。连接超时解决的是目标地址不可达的问题读取超时解决的是响应迟迟不结束的问题。两者分开设置后更容易判断请求卡在哪一步。4.3 限流与重试策略接口服务方为了保护后端资源通常会限制单位时间内的请求次数。当请求过快时可能返回 429 状态码并在响应头中给出Retry-After字段提示需要等待多久。重试不是盲目重复请求。合理的重试逻辑应该注意以下几点只对可恢复的错误重试比如超时、5xx、429不要对 400、401、403 这类参数或权限错误重试因为重试也不会成功每轮重试之间留出间隔避免造成更大压力记录每次重试的原因方便后续通过日志分析。5. 封装一个可以复用的 API 数据获取函数5.1 函数输入输出设计当取数逻辑变复杂后不要在每个脚本里复制粘贴请求代码。可以把请求、状态码检查、超时和重试封装成一个函数。函数输入至少包括 URL、请求参数、请求头、超时时间和最大重试次数输出统一为 JSON 结构遇到不可恢复错误时抛出异常。这样做的好处是调用方只关心 URL 和参数不用关心网络异常细节。后续如果要增加代理、关闭 SSL 校验、加日志只改封装函数即可不用影响下游代码。5.2 封装网络请求、超时、重试与错误分类下面给出一个可复用的请求封装示例。它把常见的requests异常做分类处理并对 429 和 5xx 做有限次数重试。import time import requests class FetchAPIError(Exception): pass def get_json(url, paramsNone, headersNone, timeout(5, 15), max_retries3): for attempt in range(1, max_retries 1): try: resp requests.get(url, paramsparams, headersheaders, timeouttimeout) if resp.status_code 429: retry_after resp.headers.get(Retry-After) wait int(retry_after) if retry_after and retry_after.isdigit() else 5 time.sleep(wait) continue resp.raise_for_status() return resp.json() except requests.exceptions.Timeout as exc: if attempt max_retries: raise FetchAPIError(f请求超时重试 {max_retries} 次后仍失败) from exc time.sleep(2 ** attempt) except requests.exceptions.HTTPError as exc: if resp.status_code 500 and attempt max_retries: time.sleep(2 ** attempt) continue raise FetchAPIError(fHTTP {resp.status_code}: {resp.text[:500]}) from exc except requests.exceptions.RequestException as exc: if attempt max_retries: raise FetchAPIError(f网络异常: {exc}) from exc time.sleep(2 ** attempt) raise FetchAPIError(请求失败已超出最大重试次数)这个封装并不复杂但已经覆盖了主要问题4xx 错误直接抛出不浪费重试5xx 错误和 429 限流做退避重试网络异常按指数间隔重试错误信息中带上响应前 500 字符方便第一时间判断问题。注意生产代码还需要处理Retry-After不是数字的情况以及记录每次重试的日志。上面的isdigit判断是简化版本。5.3 调用封装函数把结果转成 DataFrame有了封装后调用代码会变得很干净。import pandas as pd url https://jsonplaceholder.typicode.com/posts params {_limit: 20} data get_json(url, paramsparams) df pd.DataFrame(data) print(df.shape) print(df.head(2))当接口返回结构变化时错误会集中在封装函数层提示是 HTTP 状态码问题、超时问题还是响应解析问题。调用方不需要在多个脚本里各自处理一遍。6. 数据校验、落盘与后续复用6.1 拉完数据先做行数和字段校验数据到手后先校验再落盘可以避免把脏数据混入后续建模流程。一个简单的校验函数至少应检查空 DataFrame、缺字段和重复行。def validate_frame(df, required_columns): if df.empty: raise ValueError(数据为空请检查 API 响应) missing set(required_columns) - set(df.columns) if missing: raise ValueError(f缺少字段: {missing}) duplicates df.duplicated().sum() if duplicates 0: print(f警告: 存在 {duplicates} 条重复行) print(fshape: {df.shape}) print(df.dtypes)required_columns从接口文档中读取而不是凭记忆写入。对于日期字段可以先做一次pd.to_datetime转换如果抛出异常说明字段格式和预期不一致。对于数值字段可以检查是否有错误占位符比如-999、null、N/A。6.2 CSV 与 Parquet 怎么选保存数据时CSV 和 Parquet 是两种常见选择适用范围不同。对比项CSVParquet格式文本列式二进制可读性好Excel 可直接打开需要 pandas 或专业工具压缩率一般更好类型保留日期、嵌套结构容易丢保留字段类型更完整适用场景小数据、交换、肉眼检查中大批量、机器学习后续读取第 17 天的练习数据量通常不大保存为 CSV 即可。如果计划后续做更完整的特征工程建议同时保存一份 Parquet因为它在读取速度、压缩率和类型保留上都更适合机器学习流程。df.to_csv(data/raw/posts_raw.csv, indexFalse) df.to_parquet(data/processed/posts.parquet, indexFalse)6.3 原始数据、副本和校验信息不要只放在内存里内存中的 DataFrame 一旦进程结束就会消失所以落盘是必须的。更稳妥的做法是给文件名加上日期或时间戳并生成校验信息方便确认文件是否完整。import hashlib import pathlib import pandas as pd out_dir pathlib.Path(data/raw) out_dir.mkdir(parentsTrue, exist_okTrue) filename posts_20250101.csv output_path out_dir / filename df.to_csv(output_path, indexFalse) with open(output_path, rb) as f: digest hashlib.md5(f.read()).hexdigest() print(filename, digest)这份文件的哈希值可以单独记录下来。如果后续发现文件读取异常重新计算哈希后就能判断文件是否被修改或损坏。数据备份和恢复并不只是数据库层面的概念单个 CSV 文件同样需要版本管理。拉取接口后至少保留一份原始文件不要用清洗后的数据覆盖它否则出问题时只能重新调用接口。7. 常见问题和排查路径速查7.1 先按 HTTP 状态码定位问题当接口请求失败时状态码是第一排查线索。状态码含义优先检查200请求成功解析 JSON、字段是否齐全400请求参数或请求体有误参数格式、必填项、请求体大小401未认证Token 是否有效、是否过期403没有权限账号权限、接口 scope 是否声明404路径或资源不存在URL 是否正确、方法是否正确429请求过于频繁限流规则、Retry-After500 / 502 / 503服务端异常服务方状态、是否需要重试400 错误里有一个常见场景发送给模型的文本或上下文太长超过接口允许的最大长度限制。此时响应文本中通常会出现context length或token limit相关提示。处理方式是减少请求体、拆分文本或者在发送前先做 token 数估算而不是盲目重试。7.2 网络超时与连接错误怎么排查超时和连接错误看起来都像“请求不到数据”但处理方式不同。建议按以下顺序排查先确认目标地址本身可以访问可以将 URL 粘贴到浏览器或工具中验证。检查当前网络环境、代理配置和防火墙规则。检查 URL 是http还是https端口是否写对。使用短超时测试比如timeout3快速判断连接是否成功。查看错误类型ConnectTimeout通常是连不上服务器ReadTimeout通常是连接建立后响应迟迟未返回。如果服务方有状态页或文档查看是否有维护窗口或故障公告。7.3 响应结构变化导致解析失败怎么办第三方 API 的响应结构可能发生变化字段被重命名、字段类型从数字变成字符串、嵌套层级加深、甚至整个响应包了一层统一外壳。解析时不要假设所有字段都存在。data resp.json() rows data.get(items) or data # 有些接口会包一层 items更稳妥的方式是先打印响应类型和前几十个字符确认结构后写解析逻辑。使用dict.get(key, default)获取字段避免 KeyError 直接中断任务。如果解析中断跳过错误不是好的做法至少要记录异常所在的请求参数和响应片段便于还原现场。8. 工程化建议把 API 获取数据变成可监控的流程8.1 密钥、日志、监控缺一不可当脚本只是本地练习时密钥处理不规范问题不大一旦要部署到服务器或接入定时任务密钥和日志必须按工程化方式处理。密钥管理方面Token 使用环境变量或密钥管理服务不写入代码.env.example只记录键名不写真实值提交代码前检查是否意外包含了密钥文件。日志方面至少记录每次请求的 URL 标识、返回行数、耗时、失败原因、重试次数。import logging logging.basicConfig(levellogging.INFO, filenamelogs/fetch.log) logger logging.getLogger(data_api) logger.info(开始拉取数据页面: %s, page) logger.warning(请求超时重试第 %s 次, attempt) logger.error(数据校验失败缺少字段: %s, missing_fields)监控方面可以先用最简单的方案每天定时任务执行后检查输出文件是否存在、行数是否在预期范围内。当任务失败时通过日志或告警渠道收到通知。这个阶段不需要引入复杂平台先把可观测性建立起来。8.2 从一次性脚本到定时同步任务获取数据如果只需要一次放在 Jupyter Notebook 里执行没有问题。但如果数据需要每天更新就要把脚本独立出来并考虑定时调度。常见做法是把fetch_data.py设计成可命令行执行的脚本例如传入日期参数然后再接入系统的定时任务平台。任务编排时需要关心几个点任务是否幂等重复执行同一个日期任务不会产生重复数据。任务是否有超时控制超过预期执行时间要能强制结束。任务失败后是否有重跑机制是补拉前一天数据还是跳过。任务依赖关系数据拉取完成后下游清洗特征任务再启动。如果项目里已经使用工作流调度系统比如 Apache DolphinScheduler 或类似工具可以把这个 Python 脚本作为一个节点接入让数据获取任务和后续数据抽取、模型训练任务串成一条链路。即使暂时不用调度平台也应该把获取逻辑写成脚本而不是一直依赖 Notebook 手动运行。8.3 给第 17 天的落地练习清单这一天的练习如果只读不写效果会打折扣。建议按照下面的清单动手做一遍选择一个公开 API先手动请求一次记录响应结构。写出 GET 请求代码输出状态码和响应前 500 字符。把 JSON 转成 DataFrame检查行数、字段和类型。为接口加分页循环并加入最大页数保护。封装请求函数加入超时、状态码判断和重试逻辑。增加数据校验函数检查空数据和缺字段。按日期保存原始数据 CSV并记录文件哈希。尝试用保存好的 CSV 做一次简单的时间序列可视化。完成这个清单之后下一步可以把数据导入 EDA 流程比如用 pandas 做分布查看、缺失值统计和时间趋势图。API 获取只是数据链路的第一步真正体现价值的在于后续数据质量分析和特征理解。先保证数据链路稳定、可复现再谈建模顺序不要颠倒。