AWS Glue中RDD转换调用类方法触发CONTEXT_ONLY_VALID_ON_DRIVER错误

你遇到的CONTEXT_ONLY_VALID_ON_DRIVER错误核心原因是：在RDD的transformation（比如map/filter）中调用了依赖SparkSession/SparkContext的代码。

你的场景里，get_duplicates方法中执行了self.spark.createDataFrame——这个操作是必须在driver端执行的Spark API，但RDD的transformation逻辑是分发到worker节点运行的，worker无法访问driver端的SparkSession实例，因此触发错误。

哪怕你把get_duplicates改成静态方法、移到类外，只要方法内部还在调用Spark API（比如createDataFrame），问题就不会解决，因为本质是在worker端执行了driver专属的操作。

1. 重构逻辑，用DataFrame API替代RDD转换 #

Spark DataFrame的API是分布式安全的，优先用DataFrame来处理去重逻辑，避免在RDD的transformation中调用Spark API：

def main_transaction_table(self):
    # 把RDD转换为DataFrame
    df = self.spark.createDataFrame(your_rdd, schema=your_schema)
    # 用DataFrame的去重API处理，比如dropDuplicates或自定义逻辑
    deduplicated_df = df.dropDuplicates(["key_column"])
    # 后续操作直接基于DataFrame
    deduplicated_df.write.parquet("s3://your-path")

2. 让get_duplicates成为纯Python函数（无Spark依赖） #

如果必须保留RDD操作，要确保get_duplicates完全是纯数据处理逻辑，不调用任何Spark API：

# 纯Python去重函数，只处理内存中的数据
def get_duplicates(data_list):
    # 示例：按唯一键去重，纯Python实现
    unique_dict = {}
    for item in data_list:
        # 假设item是字典，用某个唯一键作为key
        key = item["transaction_id"]
        if key not in unique_dict:
            unique_dict[key] = item
    return list(unique_dict.values())

def main_transaction_table(self):
    # 在RDD转换中调用纯Python函数，无Spark依赖
    result_rdd = your_rdd.map(lambda partition_data: get_duplicates(partition_data))
    # 后续操作

3. 核心原则 #

• 所有需要SparkSession/SparkContext的操作（比如createDataFrame、read/write、广播变量创建）必须在driver端执行，不能放到RDD的map/filter/flatMap等transformation里。
• RDD的transformation逻辑只能包含纯Python代码，处理单个元素或分区内的数据，不依赖任何driver端的专属对象。

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