多线程场景下向空PySpark DataFrame追加行失败问题排查

错误原因 #

• DataFrame不可变性：PySpark DataFrame是不可变对象，unionAll（或官方推荐的union）方法不会修改原DataFrame，而是返回一个包含合并结果的新DataFrame。你的代码中init_df.unionAll(api_df)仅创建了新DataFrame，但未将其赋值回原变量，因此initial_df始终保持初始空状态。
• 线程安全问题：Spark Session和DataFrame并非线程安全对象，直接在Python线程中操作DataFrame会引发并发访问冲突，即使尝试赋值，也会因竞态条件导致数据无法正确合并。
• 低效的线程内DF操作：在每个线程中创建小DataFrame并尝试合并，不仅性能低下，还会因多线程操作Spark上下文引发未知异常。

修正方案 #

方案一：线程收集数据后统一生成DataFrame（适合中小数据量） #

放弃在线程中操作DataFrame，改为让线程返回原始数据，收集所有结果后一次性转换为符合Schema的DataFrame：

1. 修改get_json函数，返回单条记录的字典：

import json
from datetime import datetime

def get_json(booking_number: str):
    token_headers = {'Authorization': f"Bearer {token}"}
    api_response = requests.get(f'https://api_url?booking_number={booking_number}', headers=token_headers)
    json_data = json.loads(api_response.text)
    return {
        'bookingNumber': booking_number,
        'bookingProfile': json_data.get('id'),
        'json_string': api_response.text,
        'json_ingestion_time': datetime.now()
    }

2. 主函数用线程池收集数据并生成最终DataFrame：

from concurrent.futures import ThreadPoolExecutor

if __name__ == '__main__':
    spark = SparkSession.builder.appName('Raw_Check').getOrCreate()
    booking_ids = get_booking_ids(spark=spark, query='select booking_id from db.table limit 20')
    
    # 并行调用API收集数据
    results = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        for result in executor.map(get_json, booking_ids):
            results.append(result)
    
    # 转换为Spark DataFrame
    final_df = spark.createDataFrame(results, schema=table_schema) if results else prepare_empty_df(schema=table_schema)
    
    print(final_df.count())
    final_df.show()

方案二：利用Spark分布式并行处理（适合200k级大数据量） #

Spark本身是分布式计算框架，推荐使用Spark原生API实现并行调用，避免Python线程的安全问题：

import json
import requests
from pyspark.sql.functions import current_timestamp

def fetch_api_data(booking_number):
    token_headers = {'Authorization': f"Bearer {token}"}
    api_response = requests.get(f'https://api_url?booking_number={booking_number}', headers=token_headers)
    json_data = json.loads(api_response.text)
    return (booking_number, json_data.get('id'), api_response.text)

if __name__ == '__main__':
    spark = SparkSession.builder.appName('Raw_Check').getOrCreate()
    
    # 读取booking_ids为DataFrame
    booking_df = spark.sql('select booking_id from db.table limit 20')
    
    # 分布式并行调用API并生成结果
    final_df = booking_df.rdd.map(lambda row: fetch_api_data(row.booking_id))\
        .toDF(['bookingNumber', 'bookingProfile', 'json_string'])\
        .withColumn('json_ingestion_time', current_timestamp())
    
    print(final_df.count())
    final_df.show()

关键注意事项 #

• 若API有调用频率限制，需在API调用函数中添加重试、延迟逻辑，避免被封禁。
• 200k条数据优先选择方案二，Spark的分布式并行能更好利用集群资源，避免Python线程的性能瓶颈。
• 确保table_schema的字段顺序与生成的DataFrame字段顺序一致，否则createDataFrame会报错。

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