4、数据输入进阶：API 数据获取与实时数据流

上一章我们聊了文件读取和基础网络请求。说实话，那些只是开胃菜。真实世界里，数据源往往是活的——API 会变、频率会变、数据量会暴涨。我早期做过一个爬虫项目，读取静态 JSON 文件时一切正常，结果一上线对接实时 API，程序直接崩溃。嗯，原因很简单：我没做好异常处理和流式控制。

这一章，我们深入聊聊 API 的数据获取，以及怎么处理源源不断的实时数据流。这些东西，说白了就是数据工程的“基本功”。

API 数据获取：不仅仅是调用

API 调用看起来简单：发个请求，拿到 JSON，完事。但实际项目里，你会遇到无数奇怪的情况。我记得有一次对接天气 API，明明文档写着返回字段是 “temp_c”，结果实际返回的是 “temperature”。就是因为这样一个字段名差异，我们整条数据管道跑飞了三天。

我这里总结了一套相对靠谱的 API 交互流程：

  • 访问认证：最常见的就是 API Key，放在 Header 里传递。少数会用 OAuth 2.0，流程稍微复杂点。
  • 请求构建：GET 参数大多数在 URL 中传递，POST/PUT 数据则放在 Request Body 里，格式通常是 JSON。
  • 响应处理：不要只解析 JSON。一定要检查 HTTP 状态码，200 是正常，4xx 一般是请求参数问题，5xx 是服务端故障。
  • 错误重试：遇到 5xx 或网络超时，一定要自动重试，但不要马上重试，要加退避策略。
  • 限流与分页：大部分公开 API 都有速率限制，一定要在你的代码里遵守。数据多的时候，用分页遍历。

核心要点：API 调用的核心不是“能拿到数据”，而是“稳定、可控、可重试”地拿到数据。这才是专业做法。

RESTful API 实战：以 GitHub API 为例

咱们直接拿代码说话。你看下面这个例子，我用 requests 库调用 GitHub 的用户信息 API，展示了几个关键点。

import requests
import time

API_BASE = "https://api.github.com"
TOKEN = "ghp_your_token_here"  # 实际项目不要硬编码！

headers = {
    "Authorization": f"token {TOKEN}",
    "Accept": "application/vnd.github.v3+json"
}

def fetch_user(username):
    url = f"{API_BASE}/users/{username}"
    # 第一次尝试
    resp = requests.get(url, headers=headers, timeout=10)
    if resp.status_code == 200:
        return resp.json()
    elif resp.status_code in (403, 429):
        # 限流，等一会儿再试
        retry_after = int(resp.headers.get("Retry-After", 60))
        print(f"被限流了，等待 {retry_after} 秒...")
        time.sleep(retry_after)
        return fetch_user(username)  # 递归重试，注意生产环境设置递归深度
    else:
        resp.raise_for_status()

# 用法
user_data = fetch_user("torvalds")
print(user_data["name"])

你看，这个函数做了几件事：第一，设置了请求超时，防止挂死；第二，检查了状态码；第三，针对限流做了自动等待重试。我曾经在项目里忘记加超时设置，结果某个 API 挂了，整个调度系统卡了 2 小时。所以说，超时和重试是 API 交互的生命线。

分页数据的遍历策略

当 API 返回的数据量很大时（比如用户列表、订单列表），服务端通常会分页返回。GitHub 使用 Link Header 来标记下一页链接。你想想看，如果不会处理分页，你可能只能拿到前 30 条数据。

def fetch_all_repos(username):
    url = f"{API_BASE}/users/{username}/repos"
    repos = []
    page = 1
    per_page = 100  # GitHub 最大支持 100

    while True:
        params = {"page": page, "per_page": per_page}
        resp = requests.get(url, headers=headers, params=params)
        if resp.status_code != 200:
            break
        data = resp.json()
        if not data:
            break          # 没有数据了
        repos.extend(data)
        page += 1
    return repos

这个模式很简单：请求下一页，拿到数据，加到列表里，直到返回空数组。但我建议你设置一个最大页数限制，防止因为某些 bug 导致无限循环。我曾经面试过一个候选人，他的代码写得很好，就是没加这个限制，结果生产环境跑了 10 万次请求才被发现——运维部门直接找上门了。

实时数据流：从“拉”到“推”

API 调用本质上是“拉”模式：你主动问服务器要数据。但很多场景需要“推”模式——数据源一有变化，服务器自动把数据发给你。这就是实时数据流的用武之地。

目前主流的实时数据获取方式有三种：

方式 原理 适用场景 典型延迟
轮询（Polling） 客户端每隔一段时间请求一次 数据更新频率低、对延迟不敏感 几秒到几分钟
WebSocket 客户端与服务端建立长连接，双向通信 高频数据、实时推送、双向交互 毫秒级
SSE (Server-Sent Events) 服务端单向推送事件流给客户端 单向实时通知（如日志、股价、新闻） 秒级以内

我的选择建议：如果只是服务器往客户端发数据，SSE 最简单，基于 HTTP，没有复杂握手；如果需要客户端也主动发消息，那就用 WebSocket。轮询效率最低，尽量少用。

WebSocket 实战：实时数据订阅

咱们来看一个真实的 WebSocket 例子。我用 Binance 的加密货币行情 API 做演示。这个 API 会实时推送 BTC/USDT 的最新成交价。

import asyncio
import websockets
import json

WEBSOCKET_URL = "wss://stream.binance.com:9443/ws"

async def subscribe_ticker(symbol):
    url = f"{WEBSOCKET_URL}/{symbol}@trade"
    async with websockets.connect(url) as ws:
        # 连接成功后，默认会持续收到推送
        try:
            async for message in ws:
                data = json.loads(message)
                price = float(data['p'])
                qty = float(data['q'])
                event_time = data['E']
                print(f"{symbol} 成交价: {price:.2f}, 数量: {qty:.4f}, 时间: {event_time}")
                # 这里可以做实时聚合、写入数据库等操作
        except websockets.exceptions.ConnectionClosed:
            print("连接已关闭，请重连...")

# 运行
asyncio.run(subscribe_ticker("btcusdt"))

这个代码很简单：连接到 WebSocket，然后持续监听消息。每收到一条消息，就解析出价格和数量。你想想看，这是一个无限循环，程序会一直运行，直到你手动中断或连接断开。

避坑指南：WebSocket 连接可能会由于网络抖动、服务器维护等原因断开。我曾经一个项目没做重连逻辑，结果休息一个晚上回来，发现数据已经断了 6 小时。一定要加上自动重连机制。

SSE（Server-Sent Events）实战

SSE 比 WebSocket 轻量很多。它利用 HTTP 长连接，服务器持续把事件推送给客户端。Python 里用 sseclient 库非常方便。

import sseclient
import requests

SSE_URL = "https://api.example.com/events/stream"

def listen_events():
    response = requests.get(SSE_URL, stream=True)
    client = sseclient.SSEClient(response)
    for event in client.events():
        if event.event == "message":
            print(f"收到消息: {event.data}")
        elif event.event == "alert":
            print(f"重要告警: {event.data}")
        # 其他事件类型

# 启动监听
listen_events()

SSE 的 event 字段可以帮你区分不同类型的推送。这在日志聚合系统里特别有用——你可以在一个流里同时获得普通日志和错误日志，然后分别处理。

数据流处理框架：流量控制与缓冲

说到实时数据流，单靠一个 print 肯定不行。数据来得太快，你的下游系统（比如数据库、监控面板）可能处理不过来。这时候就需要流量控制和缓冲机制。

import asyncio
from collections import deque

class StreamBuffer:
    def __init__(self, max_size=1000, flush_interval=5):
        self.buffer = deque(maxlen=max_size)
        self.flush_interval = flush_interval

    async def add(self, item):
        self.buffer.append(item)
        if len(self.buffer) >= self.buffer.maxlen:
            await self.flush()

    async def flush(self):
        if not self.buffer:
            return
        # 批量发送或写入
        items_to_flush = list(self.buffer)
        self.buffer.clear()
        # 模拟写入数据库
        print(f"批量写入 {len(items_to_flush)} 条数据")
        await asyncio.sleep(0.1)  # 模拟IO

# 用法
buffer = StreamBuffer(max_size=500)
# 在 WebSocket 的回调里调用 await buffer.add(data)

核心思想：不要在每条消息来时都做一次数据库写入。攒一批，然后一次性提交。这样吞吐量能提升几十倍。我有个项目就是这样，把写入频率从每秒几百次降到了每秒十几次，数据库压力骤降。

数据输入进阶的知识结构

我把这章的核心知识点画成了一幅图，帮你理清思路。你看，整个流程从数据源出发，到最终的数据存储或下游消费，中间涉及到协议选择、错误处理、流控缓冲等关键环节。

数据输入进阶：API与实时流知识结构 数据源 协议选择 REST API (轮询) GET/POST 参数 WebSocket ws:// 双向连接 SSE 服务端推送 核心处理步骤 错误重试 · 分页 · 限流 · 缓冲 数据存储 / 下游消费 稳定、可控、可重试 是数据输入的核心原则

你看，从数据源出发，先选择协议，然后根据不同方式处理各自的细节（轮询的间隔时间、WebSocket 的重连、SSE 的事件分类），最后汇聚到核心处理层，经过错误重试、分页、限流和缓冲，最终到达数据存储或下游。这个结构，说白了就是数据输入管道的骨架。

工程化建议：日志与监控

写数据采集代码，最怕的就是“出问题了不知道”。我建议你在每一处关键节点加上日志输出：

  • 请求发出时记录一次
  • 收到响应时记录状态码和数据量
  • 重试发生时记录原因和等待时间
  • 数据最终入库时记录条数和耗时
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def safe_api_call(url, headers, max_retries=3):
    for attempt in range(1, max_retries + 1):
        logger.info(f"发起请求 [{url}], 第{attempt}次尝试")
        try:
            resp = requests.get(url, headers=headers, timeout=10)
            if resp.status_code == 200:
                logger.info(f"请求成功, 数据量: {len(resp.text)} 字节")
                return resp.json()
            elif resp.status_code in (429, 503):
                logger.warning(f"被限流或服务不可用, 状态码: {resp.status_code}")
                time.sleep(2 ** attempt)  # 指数退避
                continue
            else:
                logger.error(f"请求异常, 状态码: {resp.status_code}")
                resp.raise_for_status()
        except requests.exceptions.Timeout:
            logger.warning(f"请求超时, 等待后重试")
            time.sleep(5)
        except Exception as e:
            logger.error(f"未知错误: {e}", exc_info=True)
            break
    logger.critical(f"所有重试均失败, 放弃请求 [{url}]")
    return None

个人经验：我建议你把日志级别设置得合理些。INFO 记录正常流程，WARNING 记可恢复的异常，ERROR 记不可恢复的故障。这样日常运维查看日志时，扫一眼 WARNING 和 ERROR 就能快速定位问题。真的，省了不少救火的时间。

总结：稳定是第一位的

数据输入是整条数据管线的起点。起点不稳，后面所有环节都是白搭。API 调用要关注认证、错误处理、分页和限流；实时数据流要看重协议选择、连接管理和流量控制。把这些基本功打扎实了，后面的数据处理和可视化才能站得住脚。

← 上一章 📖 返回目录 下一章 →