接入第三方API数据——分页、限流、断点续传怎么做？

# 典型广告API采集的需求
# - 每天拉前一天所有campaign的spend/impression/click/conversion
# - 单平台账号约5000个campaign,每个返回24小时×多个细分维度
# - 单次API调用返回最多1000行,有rate limit
# - 偶发504/429,需要重试
# - 作业失败重启不能从头拉

# ❌ offset分页:数据动态变化时会出问题
def fetch_offset(api):
    page = 0
    while True:
        data = api.get('/campaigns', offset=page*100, limit=100)
        if not data: break
        yield from data
        page += 1
    # 问题: 第1页拉到后,对方插入新数据,
    # 第2页的offset=100就漏掉了原来的第100条
 
# ✓ cursor分页:基于上一页最后一条的标识
def fetch_cursor(api):
    cursor = None
    while True:
        params = {'limit': 100}
        if cursor: params['after'] = cursor
        resp = api.get('/campaigns', **params)
        yield from resp['data']
        cursor = resp.get('paging', {}).get('cursors', {}).get('after')
        if not cursor: break
    # cursor是对方返回的不透明字符串(可能是id+timestamp编码)
    # 即使数据变化,从cursor位置继续读不会错位
 
# 时间窗口分页(适合时序数据)
def fetch_time_window(api, start, end):
    cur = start
    while cur < end:
        next_t = cur + timedelta(hours=1)  # 1小时窗口
        data = api.get('/events', since=cur.timestamp(), until=next_t.timestamp())
        yield from data
        cur = next_t
    # 注意: 边界要用半开区间[start, end)避免重复

import time
from collections import deque
 
class TokenBucket:
    """令牌桶限流: 每秒rate个令牌,最多burst个"""
    def __init__(self, rate, burst):
        self.rate = rate
        self.tokens = burst
        self.last = time.time()
    def acquire(self, n=1):
        now = time.time()
        self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)
        self.last = now
        if self.tokens >= n:
            self.tokens -= n
            return True
        time.sleep((n - self.tokens) / self.rate)
        return self.acquire(n)
 
# 退避重试: 429/5xx指数退避
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(stop=stop_after_attempt(5),
       wait=wait_exponential(multiplier=2, min=4, max=300))
def call_api(url, params):
    bucket.acquire()              # 主动节流
    r = requests.get(url, params=params, timeout=30)
    if r.status_code == 429:
        # 优先尊重Retry-After header
        retry_after = int(r.headers.get('Retry-After', 60))
        time.sleep(retry_after)
        raise Exception('rate_limited')   # 触发tenacity重试
    if r.status_code >= 500:
        raise Exception(f'server_error_{r.status_code}')
    r.raise_for_status()
    return r.json()
 
bucket = TokenBucket(rate=10, burst=50)  # 10次/秒,burst 50

关键点：一定要先看Retry-After头再决定sleep多久，盲目指数退避可能等几分钟而对方实际1秒就放开了。

方案三：断点续传——持久化游标

作业可能跑几小时，中途机器挂了/网络断了/被kill了，重启不能从头来。每完成一个分页就把游标持久化到外部存储（Redis/MySQL/HDFS）。

import redis
import json
 
class CursorStore:
    """持久化游标存储,支持断点续传"""
    def __init__(self, key):
        self.r = redis.Redis()
        self.key = key
    def save(self, cursor, processed_count):
        self.r.hset(self.key, mapping={
            'cursor': json.dumps(cursor),
            'count': processed_count,
            'updated_at': time.time()
        })
    def load(self):
        data = self.r.hgetall(self.key)
        if not data: return None, 0
        return json.loads(data[b'cursor']), int(data[b'count'])
    def clear(self):
        self.r.delete(self.key)
 
def resumable_fetch(api, store):
    cursor, count = store.load()  # 从上次的位置继续
    print(f'恢复: 从cursor={cursor}继续,已处理{count}条')
    
    while True:
        try:
            resp = call_api('/campaigns', {'after': cursor, 'limit': 100})
        except Exception as e:
            print(f'失败: {e},游标已保存,下次重启从{cursor}继续')
            raise
        
        for item in resp['data']:
            write_to_kafka(item)
            count += 1
        
        cursor = resp.get('paging', {}).get('cursors', {}).get('after')
        store.save(cursor, count)  # 关键: 每页保存游标
        if not cursor: break
    
    store.clear()  # 完成后清理

方案对比

维度	offset分页	cursor分页	时间窗口分页
数据正确性	低(动态数据出错)	高	高
性能	深翻页慢	快(O(1))	快
对方支持	几乎都支持	现代API	时序数据API
断点续传	易(存offset)	易(存cursor)	易(存时间)
推荐度	✗ 仅小数据	✓ 首选	✓ 时序数据

面试高频追问

追问一：cursor过期了怎么办？

有的API的cursor有有效期（比如Meta API的cursor过24小时失效）。处理方法：① 检测cursor过期返回的特定错误码（一般是410/400），切换到时间窗口分页从上次成功的时间点继续；② 持久化游标时同时存"上次成功的最大时间戳"，cursor失效时降级用时间戳重建。这种容灾必须提前设计，不然cursor一过期作业就挂。

追问二：怎么判断数据采全了？

三种校验方法：① 总数校验——很多API有/count接口返回总条数，跟实际拉到的对比；② 时间连续性——按时间排序看有没有时间断层；③ T+1全量对账——增量拉取一周后做一次全量拉取，对比差异，长期偏差超过阈值就告警。增量+全量的混合策略是生产标准做法。

追问三：API数据怎么去重？

两层去重：① 采集层——按业务主键（比如campaign_id+date）做upsert到Kafka，相同主键覆盖；② 入仓层——MERGE INTO目标表时按主键合并，用最新的update_time版本。如果API没有稳定的更新时间戳，用server_recv_ts（采集时打的服务端时间）作为版本号。绝对不能假设"API不会返回重复数据"——网络重试就是重复的来源。

实战踩坑：API限流配额被打满影响其他业务

有次新接一个广告平台API，没看清楚"账号级别rate limit是10000/hour"是所有调用方共享的，凌晨补7天数据时把当天配额全用光了，结果业务侧实时投放系统调用同一个API全部429失败，运营那边立刻红屏。后来定了规则：所有外部API调用必须经过统一Gateway分配配额，按业务优先级分级。

# ❌ 反模式: 各业务直接调外部API,无配额管理
# 数据团队补数据 → 占满10000/hour的80%
# 业务实时投放 → 只剩20%配额,大面积429
 
# ✓ 正确: 统一Gateway分配配额
# Gateway内部维护各业务的配额池
QUOTA_POOL = {
    'business_realtime': 7000,  # 业务实时调用 70%
    'data_backfill':     2000,  # 数据补采 20%
    'data_daily':        1000,  # 数据日常 10%
}
 
@app.route('/proxy/meta_ads/<path:api_path>')
def proxy_call(api_path):
    biz = request.headers.get('X-Business-Tag')
    if not quota_check(biz, 1):
        return jsonify({'error': 'quota_exhausted'}), 429
    return forward_to_meta_api(api_path, request.args)
 
# 数据补采时,提前申请配额
# 所有API调用通过Gateway,业务隔离

解决方案：外部API配额必须做统一管理+业务分级，按优先级分配。补数据等大量调用要提前申请配额、错峰执行。这种事故出过一次后所有外部API都接到Gateway，再没出过类似问题。

总结

API采集的核心是"对方不可控"——必须主动节流、断点续传、幂等去重、定期对账。cursor分页+令牌桶限流+持久化游标+主键去重是四件套。外部API要做配额管理避免影响其他业务。面试时按"分页→限流→续传→去重→对账"五步讲，覆盖完整生产链路分高得多。

下一篇聊采集系统的可观测性：数据采集任务监控——延迟、丢失、重复指标怎么打？

你们公司接过最难调的API是哪个？踩过哪些坑？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

数据采集 面试通关手册

第16-20篇：监控一致性与架构选型

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长