PySpark分页请求遇InvalidSchema错误，如何拼接BaseURL与Next URL？

1. 修复InvalidSchema错误 #

你遇到的InvalidSchema错误是因为response.links['next']['url']返回的是相对路径（比如/api/v1/data?page=2），而非完整的带协议和域名的URL，直接传给requests.get会因缺少基础URL导致请求无效。

正确做法是用urllib.parse.urljoin安全拼接基础URL和相对路径，它会自动处理路径拼接的细节（比如基础URL末尾是否带斜杠、相对路径开头是否带斜杠）：

from urllib.parse import urljoin
base_api_url = "https://your-api-domain.com"  # 替换为你的API基础地址

# 获取下一页完整URL
next_relative_url = response.links['next']['url']
next_full_url = urljoin(base_api_url, next_relative_url)
# 发起请求
r1 = requests.get(next_full_url)

2. 打破分页死循环 #

死循环通常是因为分页终止条件缺失或逻辑错误，按以下方式修正：

• 每次请求后先检查response.links中是否存在'next'字段，不存在则终止循环
• 记录已请求过的URL，避免因API返回重复链接导致无限循环
  示例循环逻辑：

visited_urls = set()
current_url = f"{base_api_url}/initial-endpoint"  # 初始请求URL

while current_url:
    if current_url in visited_urls:
        break  # 终止重复请求
    visited_urls.add(current_url)
    
    response = requests.get(current_url)
    if response.status_code != 200:
        print(f"请求失败: {current_url}, 状态码: {response.status_code}")
        break
    
    # 处理当前页数据（比如暂存到列表或直接传入Spark）
    process_current_page(response.json())
    
    # 更新下一页URL
    if 'next' in response.links:
        current_url = urljoin(base_api_url, response.links['next']['url'])
    else:
        current_url = None

3. PySpark中分页数据的正确处理方式 #

不要在Driver端单线程循环请求（容易导致Driver负载过高，且无法利用Spark的并行能力），推荐以下两种方式：

方式一：先收集所有分页URL，再并行处理 #

适合分页数量不多的场景：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APIPagination").getOrCreate()

def fetch_and_parse(url):
    response = requests.get(url)
    return response.json() if response.status_code == 200 else None

# 第一步：收集所有有效分页URL
page_urls = []
current_url = f"{base_api_url}/initial-endpoint"
visited_urls = set()

while current_url and current_url not in visited_urls:
    visited_urls.add(current_url)
    page_urls.append(current_url)
    response = requests.get(current_url)
    if 'next' in response.links:
        current_url = urljoin(base_api_url, response.links['next']['url'])
    else:
        current_url = None

# 第二步：并行拉取并处理所有分页数据
page_rdd = spark.sparkContext.parallelize(page_urls).map(fetch_and_parse)
# 过滤无效数据
filtered_rdd = page_rdd.filter(lambda x: x is not None)
# 转为DataFrame并写入JSON
page_df = spark.read.json(filtered_rdd)
page_df.write.mode("overwrite").json("/your/output/path")

方式二：动态生成分页任务（适合大量分页） #

如果API页数极多，一次性收集所有URL会占用过多内存，可使用Spark的unfold（Spark 3.1+支持）动态生成分页任务：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, ArrayType

def get_next_page(url):
    response = requests.get(url)
    if response.status_code != 200 or 'next' not in response.links:
        return (response.json(), None)
    next_url = urljoin(base_api_url, response.links['next']['url'])
    return (response.json(), next_url)

# 初始URL作为种子
initial_df = spark.createDataFrame([(f"{base_api_url}/initial-endpoint",)], ["url"])

# 动态展开分页
page_data_df = initial_df.unfold(
    "url",
    get_next_page,
    "page_data"
).select("page_data")

# 写入JSON
page_data_df.write.mode("overwrite").json("/your/output/path")

4. 排查无结果问题 #

如果代码运行无结果，按以下步骤排查：

• 打印每次请求的URL和响应状态码，确认请求是否成功
• 打印response.json()的内容，确认API返回的数据是否为空
• 检查Spark任务的日志，查看是否有数据过滤、转换环节的错误
• 确认输出路径的权限，确保Spark有权限写入目标位置

内容的提问来源于stack exchange，提问作者Arun.K