PySpark打印RDD触发RuntimeError：已创建SparkSession仍报错

问题根源 #

• Spark SQL的udf()生成的是DataFrame专用函数，不能直接在RDD的map操作里调用。UDF依赖SparkSession上下文，而RDD的map是在分布式executor节点执行，直接调用UDF会导致上下文缺失，触发报错。
• 你的cap_title函数本身还有逻辑错误：x = words.split('.')后，列表推导式遍历的是原字符串words的每个字符，还错误调用了列表x的capitalize()，这会导致后续运行报错，得先修正这个逻辑。

解决方案 #

方案1：直接用普通Python函数处理RDD #

既然是操作RDD，直接用你定义的Python函数（先修正逻辑），不需要转成UDF：

# 修正函数逻辑：按空格拆分标题，每个单词首字母大写（如果是按点拆分就把split(' ')改成split('.')）
def cap_title(words):
    split_words = words.split(' ')
    capitalized_words = [word.capitalize() for word in split_words]
    return ' '.join(capitalized_words)

# 直接在RDD的map里调用普通Python函数，不用UDF
transformed_rdd = c_rdd.map(lambda x: [x[0], cap_title(x[1])])
print(transformed_rdd.take(5))

方案2：转成DataFrame用UDF处理 #

如果想用UDF，先把RDD转成DataFrame，再用UDF操作：

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, StructType, StructField

# 确保SparkSession已创建
spark = SparkSession.builder.appName("CapitalizeTitle").getOrCreate()

# 修正函数逻辑
def cap_title(words):
    split_words = words.split(' ')
    capitalized_words = [word.capitalize() for word in split_words]
    return ' '.join(capitalized_words)

# 创建UDF
capitalize_udf = udf(cap_title, StringType())

# 把RDD转成DataFrame，假设c_rdd的结构是(id, title)
schema = StructType([
    StructField("id", StringType(), True),
    StructField("title", StringType(), True)
])
df = spark.createDataFrame(c_rdd, schema)

# 应用UDF并打印结果
transformed_df = df.withColumn("capitalized_title", capitalize_udf(df["title"]))
transformed_df.select("id", "capitalized_title").show(5)

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