调用API时持续出现数据缺失问题的技术咨询

日常通过API获取每日交易数据集，每日包含4万余条记录，单条记录含500个参数，返回JSON文件最大可达90MB。目前调用API持续出现数据缺失问题，缺失数据多集中在日间时段，但无固定规律：既非每日同一时间出现，也非达到特定交易数量阈值后触发。若因大小或记录数限制，理应缺失文件末尾数据，而非中间片段。因不确定是否被允许访问该数据，不便询问API提供方。已部署2个独立Function App，每日合计调用API 10次，试图通过多请求补全数据，每日结束后合并JSON并去重，但仍无法实现100%数据覆盖，仍存在缺失。

# Retrieve environment variables
API_TOKEN = os.environ.get("API_TOKEN")
BASE_URL = os.environ.get("ROOT_URL")
STORAGE_CONNECTION_STRING = os.environ.get("AzureWebJobsStorage")

if not all([API_TOKEN, BASE_URL, STORAGE_CONNECTION_STRING]):
    logging.error("Missing required environment variables.")
    return

endpoint = "/XXXXX" #removed for privacy 
api_url = f"{BASE_URL}{endpoint}"
headers = {'Authorization': f'Basic {API_TOKEN}'}
params = {
    'fromDate': today,
    'toDate': today,
    'page': 1,
    'pageSize': 500
}

# Azure Blob Storage setup
blob_service_client = BlobServiceClient.from_connection_string(STORAGE_CONNECTION_STRING)
container_name = "api-results"
json_blob_name = f"{today}_{current_hour}.json"
json_blob_client = blob_service_client.get_blob_client(container=container_name, blob=json_blob_name)

all_data = []
total_fetched = 0

try:
    # Check if JSON blob already exists and load existing data
    if json_blob_client.exists():
        existing_blob = json_blob_client.download_blob().readall()
        try:
            existing_data = json.loads(existing_blob.decode("utf-8"))
            if isinstance(existing_data, list):
                all_data.extend(existing_data)
            logging.info(f"Loaded existing JSON data: {len(existing_data)} records.")
        except json.JSONDecodeError:
            logging.warning("Failed to decode existing JSON, starting fresh.")

    # Fetch API data page-by-page
    while True:
        response = requests.get(api_url, headers=headers, params=params, timeout=15)
        logging.info(f"API Response: {response.status_code}, Page: {params['page']}")

        if response.status_code != 200:
            logging.error(f"API call failed with status: {response.status_code}")
            break

        data = response.json()
        if not data:
            logging.warning(f"No data returned from API for {today} on page {params['page']}")
            break

        all_data.extend(data)
        total_fetched += len(data)
        logging.info(f"Fetched {len(data)} rows from page {params['page']} (Total so far: {total_fetched}).")

        # Save JSON immediately
        try:
            formatted_json = json.dumps(all_data, indent=2)
            json_blob_client.upload_blob(formatted_json, overwrite=True, content_settings=ContentSettings(content_type='application/json'))
            logging.info(f"JSON saved successfully in Blob Storage: {json_blob_client.url}")
        except Exception as e:
            logging.error(f"Failed to upload JSON to Blob Storage: {str(e)}")

        # Handle pagination
        has_next = False
        if 'X-Pagination' in response.headers:
            try:
                pagination = json.loads(response.headers['X-Pagination'])
                total_records = pagination.get('TotalRecords', 'Unknown')
                has_next = pagination.get('HasNext', False)
                logging.info(f"Pagination Info: {pagination}")
            except json.JSONDecodeError:
                logging.warning("Failed to parse X-Pagination header.")

        if not has_next:
            logging.info(f"All pages fetched. Total records retrieved: {total_fetched} / {total_records}.")
            break
        params['page'] += 1

except requests.RequestException as e:
    logging.error(f"Error while calling API: {str(e)}")

1. 是否存在常见的API行为或响应限制可解释此类数据缺失现象？ #

• 动态限流/负载保护：日间交易高峰时API服务端负载过高，可能静默丢弃部分请求或返回不完整数据，且不返回错误码，避免触发客户端重试进一步加重负载。
• 数据快照一致性问题：API返回的是实时生成的数据集快照，日间交易频繁，生成快照时可能跳过正在写入的交易记录，导致中间时段数据缺失。
• 分页逻辑隐藏规则：若API使用偏移量分页，服务端数据实时更新时，新插入的记录会导致后续分页偏移量失效，跳过部分原有记录；部分API基于时间窗口分页，窗口边界的记录可能被漏取。
• 网络层丢包/截断：日间网络拥堵，请求或响应数据包丢失，requests默认无重试机制，导致部分分页数据未完整接收；服务端处理超时可能返回部分数据但仍标记200状态码，客户端误判数据完整。
• 权限/数据范围隐性限制：可能存在未明确告知的权限限制，比如日间部分敏感交易数据被过滤，或API对单用户的日请求数据量有隐性上限，超出后随机丢弃部分数据。

2. 针对该场景，有哪些可靠策略可确保完整的数据捕获？ #

• 基于唯一ID的校验补全：提取每条交易记录的唯一标识（如交易ID），每次获取后对比已知的总记录数（从分页头或历史数据估算），识别缺失的ID范围，定向发起补拉请求。
• 缩小请求时间粒度：将按天请求改为按小时/15分钟的时间窗口请求，减少单次请求的数据量，降低服务端处理压力和出错概率，同时便于定位缺失的具体时段。
• 添加智能重试机制：对每个分页请求实现指数退避重试（如用tenacity库），覆盖网络超时、非200状态码、返回数据量异常等场景，避免单次请求失败导致数据缺失。
• 切换游标分页：若API支持，改用基于最后一条记录ID或时间戳的游标分页替代偏移量分页，避免因数据实时更新导致的偏移量失效问题。
• 多请求交叉验证：从不同Function App发起不同时间窗口的请求，合并后去重，再对比各时段的记录数分布，对异常少数据的时段再次发起补取请求。
• 本地缓存已获取ID：维护一个已获取的交易ID集合，每次请求前过滤掉已存在的ID（若API支持参数过滤），减少重复数据，同时快速识别缺失ID。

3. 可通过哪些日志记录或测试手段排查问题根源？ #

• 增强日志维度：记录每个分页请求的详细信息：请求时间、响应耗时、返回记录数、分页头完整内容、当前页首尾记录的唯一ID；每日合并后统计缺失的ID范围及对应时段，关联日志分析。
• 模拟高峰测试：在日间手动触发多次请求，对比不同时间点的响应记录数和完整性；或在测试环境模拟高并发请求，复现可能的服务端限流场景。
• 数据一致性校验：每日获取完成后，统计各小时段的记录数，与历史均值对比，定位异常时段；同时对比同一时间窗口多次请求的记录差异，找出缺失的具体分页。
• 抓包分析：用tcpdump或Wireshark抓取日间API请求的数据包，检查响应是否完整、是否存在数据包丢失或截断情况，验证是否为网络层问题。
• 单页重复性测试：对同一分页连续发起多次请求，对比返回的记录是否一致，判断服务端是否存在返回不稳定或随机丢数据的情况。

内容的提问来源于stack exchange，提问作者Noob2025