刚接触机器学习时绝大多数教程会告诉你先下载一份 CSV 数据然后pd.read_csv()交给模型训练。这个方法本身没错但一旦进入真实业务场景你会发现数据从来不是“一个文件”那样听话。它可能躺在公司数据库里可能来自第三方开放平台也可能由多个业务系统实时产生。如果只能学会处理静态文件你的数据管道在第一步就会卡住。这一篇对应 CampusX 100 天机器学习系列的第 17 天主题是“从 API 获取数据”。我的核心判断是API 是你和实时数据世界之间的桥梁掌握它不代表你只是会调用一个接口而是意味着你具备了把外部数据接入机器学习流程的完整能力。从环境搭建、REST 基础、requests调用、JSON 解析到分页处理、异常重试和最终构建成训练集我们会把这些步骤完整串起来。读完你可以直接照着做而不是只记住几个 API 名词。1. 为什么机器学习项目离不开 API 数据获取很多人误以为机器学习工作流是从“拿到一份标注好的数据集”开始的。事实恰恰相反真实项目中数据获取往往占据整个项目 50% 以上的时间。这里的“获取”不只是下载文件还包括从多个数据源抽取、合并、校验和落库。而 API 正是其中最标准的接入方式。如果说数据库是公司内部的数据仓库那么 API 就是对外提供数据服务的“窗口”。第三方天气服务、股票行情、社交平台、物联网设备上报几乎都以 API 形式开放数据。你不需要知道对方的数据存在哪台服务器上不需要直接连接对方的数据库只需要根据约定好的接口文档发起 HTTP 请求就能拿到结构化数据。这种解耦设计让跨团队、跨公司的数据协作成为可能。对机器学习而言API 获取数据还有一层现实意义很多模型需要“新鲜”数据。比如一个销量预测模型训练时可能用一年前到昨天的历史订单但到了线上推理阶段模型需要读取当天甚至当前小时的订单数据。这时候数据源往往是业务系统的 API而不是某份训练时导出的静态文件。换句话说API 不只是训练阶段的数据入口也是推理阶段实时特征的数据入口。从 100 天学习路线的角度第 17 天放在这个位置很合理。前面几天你掌握了 Python 基础、NumPy、Pandas 和可视化这些能力处理的是“已经拿到的数据”而从今天开始你需要解决“数据从哪里来”的问题。只有把数据获取补上后续的特征工程、模型训练和部署才算完整闭环。2. 从 API 获取数据的基本概念2.1 什么是 APIAPI 全称 Application Programming Interface中文一般翻译为“应用程序编程接口”。你可以把它理解成两个软件系统之间约定的“服务窗口”调用方发出一个格式化请求提供方返回一个格式化响应。在机器学习场景中最常见的 API 类型是 REST API。REST 不是一套硬性协议而是一种基于 HTTP 的接口设计风格它约定通过 URL 定位资源。通过 HTTP 方法表达操作。通过状态码表示结果。通过 JSON或 XML传递数据。2.2 HTTP 方法与请求-响应模型当你在浏览器地址栏输入一个网址并回车浏览器执行的就是一次 HTTP GET 请求。API 调用同样如此requests库会替你把请求发送到目标服务器并接收响应。不同 HTTP 方法表达不同语义方法作用机器学习中的常见场景GET读取数据拉取训练数据、查询特征数据POST提交数据或触发复杂查询调用推理服务、提交批量测试样本PUT / PATCH更新资源更新数据标注或配置DELETE删除资源清理旧数据较少直接使用在获取数据阶段你 90% 的时间只使用 GET 和 POST。GET 适合简单的数据拉取POST 适合参数复杂、数据量大的查询或者调用远程模型服务。2.3 JSONAPI 数据的主流格式JSONJavaScript Object Notation是一种轻量级数据交换格式现在的 Web API 几乎默认返回 JSON。一个典型的 API 返回结果长这样{ code: 200, data: [ {id: 1, title: 机器学习, score: 0.98}, {id: 2, title: 数据分析, score: 0.87} ], message: success }Python 的requests库可以直接把响应体解析为字典和列表组成的 Python 对象随后用pandas.DataFrame()轻松转成表格结构。JSON 与 Python 字典的相似性是 Python 成为数据获取首选语言的重要原因。2.4 API 认证与权限不是所有 API 都允许匿名访问。很多数据平台要求调用请求携带身份凭证常见认证方式是 API Key在请求头中附加Authorization: Bearer api_key或把 key 作为查询参数传递。对学习者来说请记住一个重要原则API Key 等同于账户密码绝不能提交到公共仓库。在本地练习时建议从环境变量或本地配置文件中读取而不是硬编码在脚本里。3. 环境准备与依赖安装这一节我们搭建一个最小可用的 Python 环境。无论你使用本机 Python、Anaconda 还是虚拟环境工具目标都是一致的安装requests和pandas。3.1 确认 Python 版本示例代码适合 Python 3.8 及以上版本。可以先在终端执行python --version如果你的机器上同时存在多个 Python 版本建议使用虚拟环境隔离项目依赖。3.2 创建虚拟环境推荐# 创建虚拟环境 python -m venv ml_env # 激活虚拟环境Windows ml_env\Scripts\activate # 激活虚拟环境macOS / Linux source ml_env/bin/activate虚拟环境的目的是让每个项目的依赖互不干扰。机器学习项目尤其如此因为你可能会同时使用不同版本的库。3.3 安装依赖pip install requests pandas依赖安装完成后验证导入是否正常import requests import pandas as pd print(requests.__version__) print(pd.__version__)输出两个版本号说明环境搭建成功。版本号本身不作为本文硬性依赖重点是保持环境干净、可复现。4. 第一个 API 请求requests 库初体验环境准备好了我们先写一个最小示例从公开 API 中拉取一组数据。这里使用的是教学常用的 JSONPlaceholder 假数据 API无需注册即可访问响应结构清晰适合用来理解请求-响应流程。# 文件路径api_demo/first_request.py import requests url https://jsonplaceholder.typicode.com/posts response requests.get(url) # 状态码是 200 表示请求成功 print(HTTP 状态码:, response.status_code) # 将响应体解析为 Python 对象 data response.json() print(返回数据类型:, type(data)) print(返回数据条数:, len(data)) print(第一条数据:, data[0])运行这段代码你会看到类似输出HTTP 状态码: 200 返回数据类型: class list 返回数据条数: 100 第一条数据: {userId: 1, id: 1, title: ..., body: ...}这段示例虽短但包含了核心三步构造请求、接收响应、解析 JSON。我记得很多初学者第一次写 API 调用时会直接拿response.text做字符串切片这是不必要的。response.json()是最直接的方式。如果只返回单一对象解析后的data就是一个字典如果返回列表data就是一个列表。在不确定结构时先在 Jupyter Notebook 或终端里打印type(data)是最有效的检查手段。这个 API 还支持传入查询参数来做简单的数据过滤。如果只想获取前 5 条数据params {_limit: 5} response requests.get(url, paramsparams) limited_data response.json() print(限制后的条数:, len(limited_data))真正值得注意的一点是不要手动把参数拼在 URL 字符串里除非你需要调试 URL 编码。requests的params参数会替你做 URL 编码避免中文或特殊字符引发问题。5. 将 API 响应转换为机器学习数据集拿到 JSON 只是第一步。机器学习需要的是结构化的二维表格所以下一步是把列表类型的 JSON 转换成 Pandas DataFrame。这一步是整个环节里最直观也最容易被低估的部分。5.1 简单列表转 DataFrame如果 API 返回的是一个对象列表转换非常直接import requests import pandas as pd url https://jsonplaceholder.typicode.com/posts response requests.get(url) data response.json() df pd.DataFrame(data) print(df.head()) print(df.info())df.info()会告诉你每一列的数据类型、非空值数量等信息。做机器学习前的数据检查几乎都会从这里开始。5.2 嵌套 JSON 的处理真实 API 的返回结构往往不是扁平列表。比如数据可能被包在一个data字段里{ status: ok, data: [ {id: 1, value: 100}, {id: 2, value: 200} ] }这时需要先提取data字段再做DataFrame转换response requests.get(url) json_body response.json() records json_body[data] df pd.DataFrame(records)更麻烦的情况是字段本身是嵌套字典或列表。例如一条记录里有user字段下面还包含name和agerecords [ {id: 1, user: {name: Alice, age: 30}}, {id: 2, user: {name: Bob, age: 25}} ] # 方法一先展开再转换 flat_records [ { id: item[id], name: item[user][name], age: item[user][age] } for item in records ] df pd.DataFrame(flat_records) print(df)你可以在这一步根据业务需求做字段筛选、重命名和类型转换。对于机器学习这里就是特征工程的起点哪些字段适合做特征哪些字段包含目标值哪些字段需要丢弃都在此时决定。5.3 把获取函数封装起来为了后续复用建议把“获取数据并转成 DataFrame”的逻辑封装成一个函数def fetch_posts_to_dataframe(limit: int 100) - pd.DataFrame: 从 API 获取文章数据并转为 DataFrame。 url https://jsonplaceholder.typicode.com/posts params {_limit: limit} response requests.get(url, paramsparams, timeout10) response.raise_for_status() return pd.DataFrame(response.json())这里有两个值得学习的工程细节timeout10防止请求永远挂起response.raise_for_status()在状态码非 200 时主动抛出异常。很多初学者省略这两项等到程序在真实网络环境里无响应才意识到问题有多难排查。6. 分页与批量拉取的完整实现真实业务 API 几乎都有返回数量上限不会一次性返回所有数据。比如一次最多返回 100 条记录而你有 100 万条数据要拉取。这时必须处理分页。常见的分页方式有两种分页方式参数示例特点页码分页page1limit100简单直观API 性能压力相对可控游标分页cursorxxxx更适合数据频繁新增和删除的场景内容一致性更好以页码分页为例核心思路是在循环中逐页请求直到拿不到新数据为止# 文件路径api_demo/fetch_all.py import time import requests import pandas as pd def fetch_all_posts(max_pages: int 10) - pd.DataFrame: 循环获取所有文章数据。 参数: max_pages: 最大请求页数防止意外死循环 url https://jsonplaceholder.typicode.com/posts page 1 all_records [] while page max_pages: params {_page: page, _limit: 50} response requests.get(url, paramsparams, timeout10) if response.status_code ! 200: print(f第 {page} 页请求失败状态码{response.status_code}) break records response.json() if not records: break all_records.extend(records) print(f已获取第 {page} 页本页 {len(records)} 条) page 1 # 避免请求频率过高触发服务端限流 time.sleep(0.3) return pd.DataFrame(all_records) if __name__ __main__: df fetch_all_posts() print(总计获取条数:, len(df)) print(df.head())这个函数里有几个判断值得留意max_pages是安全上限。如果因为接口文档理解错误导致分页条件永远满足程序也可能无限循环设置上限是必要的保护。if not records: break是结束循环的出口。当 API 返回空列表说明没有更多数据了。time.sleep(0.3)是对 API 提供方的尊重。控制请求频率避免被封禁。失败时打印日志而不是直接raise更适合批量补数据的场景。你可以记录失败的页码等后续重试。运行后你会看到每页的获取日志最终汇总成一个 DataFrame。这就是一个最小可用的批量数据采集器。7. 为数据获取环节加上容错机制真实网络环境里请求失败几乎是必然的Wi-Fi 波动、服务端临时过载、接口限流、JSON 格式变化都可能让程序中断。所以批量获取数据时容错设计和数据获取本身同等重要。7.1 使用 Session 提升性能如果多次请求同一个 API建议使用requests.Session()。Session 会复用底层 TCP 连接减少握手开销速度明显优于每次新建连接。import requests session requests.Session() session.headers.update({Accept: application/json}) url https://jsonplaceholder.typicode.com/posts for page in range(1, 4): response session.get(url, params{_page: page, _limit: 10}, timeout10) print(page, response.status_code)7.2 自动重试机制使用requests配合urllib3的Retry可以实现网络错误和服务器错误状态码的自动重试import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_session_with_retry( retries: int 3, backoff_factor: float 1.0, status_forcelist: tuple (500, 502, 503, 504) ) - requests.Session: 创建带自动重试机制的 Session。 backoff_factor 用于计算重试等待时间 等待时间 backoff_factor * (2 ** (重试次数 - 1)) session requests.Session() retry_strategy Retry( totalretries, backoff_factorbackoff_factor, status_forceliststatus_forcelist, allowed_methods[GET, POST] ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session session create_session_with_retry() url https://jsonplaceholder.typicode.com/posts try: response session.get(url, params{_page: 1, _limit: 5}, timeout10) print(请求成功:, response.status_code) except requests.exceptions.RequestException as e: print(请求最终失败:, e)注意自动重试不适合 POST 等非幂等请求。POST 可能产生写入操作重试可能导致数据重复提交。上面的allowed_methods[GET, POST]仅作为教学示例实际生产环境请谨慎放开 POST 的自动重试。7.3 数据校验与落盘数据拉到本地后应该立即做两层校验数量校验实际获取的行数是否等于预期行数。关键字段校验必要字段是否存在是否有全空列。校验通过后尽快将原始响应保存到本地。建议保留一份“原样”数据而不是只保存清洗后的数据。原始数据是排查问题时的唯一依据df.to_csv(raw_posts.csv, indexFalse, encodingutf-8)如果需要保留 JSON 原始结构也可以直接存 JSON Linesimport json with open(raw_posts.jsonl, w, encodingutf-8) as f: for record in all_records: f.write(json.dumps(record, ensure_asciiFalse) \n)8. 常见问题与排查思路API 数据获取的报错信息看似五花八门但大多数问题的根因是固定的。整理一份高频问题清单遇到问题时按表排查会高效很多。问题现象可能原因排查方式解决方案ConnectionError网络不通、域名解析失败、对方服务不可达检查网络连通性尝试在浏览器访问该 URL确认网络环境增加timeout必要时联系接口提供方一直请求无响应对方服务器过慢或触发了防火墙策略抓包查看请求发出状态查看程序是否卡在某一行设置timeout10加入自动重试机制HTTP 401 UnauthorizedAPI Key 缺失、错误或已过期打印请求头确认是否携带了认证信息从环境变量重新读取密钥重新申请 API KeyHTTP 403 Forbidden权限不足、IP 被限制或请求频次过高检查接口文档的权限说明查看服务端错误响应体降低请求频率确认账号权限范围HTTP 404 Not FoundURL 路径拼写错误、资源不存在对比接口文档中的完整请求路径修正 URL注意大小写和斜杠HTTP 429 Too Many Requests请求频率超过限流阈值查看响应头中的Retry-After字段增加休眠时间实现指数退避JSONDecodeError服务端返回的不是 JSON或被反代页面拦截打印response.text[:500]查看实际内容确认响应格式检查请求头Acceptpandas.errors.ParserError将非列表 JSON 直接传给DataFrame判断响应最外层是字典还是列表先提取data字段再转 DataFrame中文乱码编码格式不统一检查响应头中的Content-Type字符集使用.encoding或手动指定utf-8解码数据量远小于预期分页条件错误或 URL 参数大小写不匹配打印每页返回条数检查params拼写逐页调试先单独请求一页确认返回条数如果你遇到的报错不在表里记住一个通用排查顺序先看状态码再看响应体最后看代码逻辑。状态码告诉你服务端是否正常响应体告诉你具体报错原因代码逻辑定位你的参数或解析方式是否正确。大多数人一开始就翻代码反而忽略了最有价值的响应体信息。9. 最佳实践与工程建议从能够“跑通”到能够“稳定运行”中间隔着一系列工程习惯。这些习惯不会在单次 API 调用里体现但在数据量大、请求频繁、运行时间长的项目中至关重要。9.1 配置管理密钥绝不能写死在代码里API Key、Token、数据库连接串都属于敏感配置。推荐在项目根目录创建.env文件并使用环境变量读取# 文件路径.env请加入 .gitignore API_KEYyour_api_key_here API_BASE_URLhttps://api.example.comPython 侧可以使用os.getenv读取import os api_key os.getenv(API_KEY) if not api_key: raise RuntimeError(缺少环境变量 API_KEY请检查 .env 文件)把.env加入.gitignore避免密钥不小心提交到 Git 仓库。9.2 幂等设计重复执行不会产生脏数据批量采集脚本应该支持重复执行。每次运行前建议记录本次运行的时间戳和请求参数写入数据库时使用唯一键或时间窗口去重。这样即使中途失败后重新运行也不会因为重复写入造成数据膨胀。9.3 日志与监控不要只依赖print()。数据量大时把运行日志写入文件方便事后回溯import logging logging.basicConfig( filenamefetch_data.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logging.info(开始获取数据参数%s, params) logging.error(请求失败页面: %s错误: %s, page, exc_infoTrue)9.4 控制请求频率无明显限流的 API单线程场景下建议请求间隔至少 200 毫秒到 1 秒。如果接口文档明确给出了限流阈值比如每分钟 60 次你应该在代码里主动计算并控制而不是依赖随机 Sleep。9.5 本地缓存如果数据不是每分钟都变化建议把当天获取的结果缓存为 CSV 或 Parquet 文件。后续分析直接读取本地缓存既减少对 API 的请求压力也避免重复拉取产生的不确定性。import os cache_path cache/2025-01-01_posts.parquet if os.path.exists(cache_path): df pd.read_parquet(cache_path) print(读取本地缓存) else: df fetch_all_posts() os.makedirs(cache, exist_okTrue) df.to_parquet(cache_path, indexFalse) print(从 API 获取并写入缓存)9.6 生产环境的部署考虑如果这个数据获取流程要部署到服务器或定时任务中建议额外考虑使用调度工具如 cron、Airflow、DolphinScheduler 等定期运行将原始数据写入对象存储或数据库对失败任务设置告警每个采集任务有独立的日志和可观测指标。这些内容超出第 17 天的主线但属于从学习走向工程的必经之路。10. 总结与后续学习方向今天我们完成了一条完整的数据获取链路理解 REST API 的基本概念用requests发起请求解析 JSON 响应把结果转成 Pandas DataFrame处理分页和异常最后将清洗后的数据落地为文件。对应 CampusX 100 天机器学习系列第 17 天你已经具备了继续学习特征工程和模型训练的数据基础。这一天的内容真正重要的事情不在于requests.get()本身而在于你要建立一个意识真实世界的数据是动态的、分散的、需要协议才能接入的。API 就是那个协议。以后无论你面对的是哪个平台的数据第一步都是读懂接口文档然后按照文档构造请求、处理响应、管理频率。下一步建议你找一两个真实且免费的公开 API 练手尝试自己写一个包含分页、重试和落盘的完整采集脚本。可以选天气历史数据、公共财经行情或地区开放数据集关键是找响应结构相对规整、文档清晰的接口。当你亲手把一个“只有接口文档”的数据源变成可以训练的 DataFrame 时第 17 天的内容才算真正消化了。之后的第 18 天、第 19 天大概率会进入数据清洗、合并和可视化等主题。到那时你会发现今天打下的数据获取基础会让你在后续所有数据相关环节中都走得更顺。建议把今天这篇文章收藏备用等到真正开始写采集脚本时再对照执行一遍。