如何在AWS EMR的PySpark任务中仅调用一次第三方API并共享至所有executors

方案1：Driver端调用API + 广播变量（Broadcast Variable） #

这是最常用的方案，核心逻辑是仅在Driver节点调用一次API，然后通过Spark的广播变量将映射数据分发到每个executor节点的内存中，该节点上的所有Task都会复用这份数据，不会重复调用API。

代码示例 #

import requests
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 初始化SparkSession
spark = SparkSession.builder.appName("API_Map_Aggregation").getOrCreate()

# Driver端单独调用第三方API获取映射关系
def fetch_mapping():
    api_url = "https://your-third-party-api.com/mapping-endpoint"
    try:
        # 按需添加请求头、参数等
        response = requests.get(api_url, timeout=15)
        response.raise_for_status()  # 触发HTTP错误
        return response.json()  # 假设返回格式为 {"key": "value", ...} 的字典
    except Exception as e:
        print(f"API调用失败: {str(e)}")
        raise  # 终止任务，避免后续无效执行

# 获取映射并广播
mapping_dict = fetch_mapping()
broadcast_mapping = spark.sparkContext.broadcast(mapping_dict)

# 定义UDF，使用广播变量做映射
def map_column_value(input_key):
    # 处理key不存在的情况，返回默认值
    return broadcast_mapping.value.get(input_key, "UNKNOWN")

map_udf = udf(map_column_value, StringType())

# 读取原始数据，新增映射列
raw_df = spark.read.parquet("s3://your-input-data-path/")
processed_df = raw_df.withColumn("new_mapped_column", map_udf(raw_df["source_column"]))

# 执行聚合任务
aggregated_df = processed_df.groupBy("new_mapped_column").count()

# 输出结果到目标路径
aggregated_df.write.mode("overwrite").parquet("s3://your-output-data-path/")

# 任务结束后手动释放广播变量（可选，Spark会自动清理）
broadcast_mapping.unpersist()
spark.stop()

优势 #

• 严格保证仅调用一次API，节省API配额且避免数据不一致
• 广播变量会自动优化分发逻辑（比如通过BitTorrent式传输），减少网络开销
• 映射数据存储在executor内存中，Task访问速度快

方案2：Driver端调用API + 共享存储（HDFS/S3） #

如果API返回的映射数据量极大（比如百万级以上），广播变量可能会占用过多executor内存，此时可以将映射数据写入EMR集群可访问的共享存储（HDFS或S3），让每个executor仅读取一次并缓存到本地。

代码示例 #

import requests
import json
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import subprocess

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

# Driver端调用API获取映射
def fetch_mapping():
    api_url = "https://your-third-party-api.com/mapping-endpoint"
    try:
        response = requests.get(api_url, timeout=15)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"API调用失败: {str(e)}")
        raise

mapping_dict = fetch_mapping()

# 将映射写入HDFS的共享路径
hdfs_mapping_path = "/user/hadoop/mapping.json"
# 先写入Driver本地，再上传到HDFS
with open("/tmp/local_mapping.json", "w") as f:
    json.dump(mapping_dict, f)
subprocess.run(["hdfs", "dfs", "-put", "-f", "/tmp/local_mapping.json", hdfs_mapping_path], check=True)

# 在executor端加载映射（用lazy缓存避免重复读取）
mapping_cache = {}
def load_and_cache_mapping():
    if not mapping_cache:
        # 从HDFS读取映射文件到本地
        subprocess.run(["hdfs", "dfs", "-get", hdfs_mapping_path, "/tmp/local_mapping_copy.json"], check=True)
        with open("/tmp/local_mapping_copy.json", "r") as f:
            mapping_cache.update(json.load(f))
    return mapping_cache

# 定义UDF使用缓存的映射
def map_column_value(input_key):
    return load_and_cache_mapping().get(input_key, "UNKNOWN")

map_udf = udf(map_column_value, StringType())

# 后续数据处理、聚合逻辑同方案1
raw_df = spark.read.parquet("s3://your-input-data-path/")
processed_df = raw_df.withColumn("new_mapped_column", map_udf(raw_df["source_column"]))
aggregated_df = processed_df.groupBy("new_mapped_column").count()
aggregated_df.write.mode("overwrite").parquet("s3://your-output-data-path/")

spark.stop()

适用场景 #

• 映射数据量极大，广播变量会导致executor内存不足
• 需要长期复用这份映射数据（后续任务可直接读取存储中的文件，无需再次调用API）

关键注意事项 #

1. API调用容错：必须在Driver端处理API调用的异常（超时、HTTP错误等），避免任务在执行阶段才发现API失败
2. 默认值处理：UDF中要对不存在的key设置默认值，防止抛出KeyError导致Task失败
3. 存储权限：确保所有executor节点有权限访问共享存储（HDFS/S3）的路径
4. 数据时效性：本方案仅适用于任务执行期间映射数据无需更新的场景，如果映射会动态变化，需考虑添加缓存失效机制

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