PySpark如何转换列类型以匹配DataFrame union合并的类型要求

报错的核心原因是你初始化空DataFrame时，将probability字段定义为DoubleType，但PySpark机器学习模型的transform输出的probability字段是MLlib专属的向量类型VectorUDT，二者类型不兼容导致union失败。

方案1：对齐空DataFrame的schema与模型输出一致 #

如果你需要保留完整的概率向量，直接修改初始schema的字段类型即可，代码如下：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType
from pyspark.ml.linalg import VectorUDT # 导入MLlib向量类型

sc = SparkContext.getOrCreate()
spark = SparkSession(sc) 

# 修改probability字段类型为VectorUDT
schema = StructType([
    StructField("row_num",IntegerType(),True),
    StructField("label",IntegerType(),True),
    StructField("probability",VectorUDT(),True),
])
empty = spark.createDataFrame(sc.emptyRDD(), schema)

# 后续模型预测、union逻辑不变
model = LogisticRegression().fit(train_data)
preds = model.transform(test_data)
all_preds = empty.unionAll(preds)

方案2：把模型输出的概率向量转为Double类型 #

如果你只需要提取单类的概率值（比如二分类场景下取正例的概率），可以对预测结果的probability字段做转换，匹配你原来的空表schema：

from pyspark.sql.functions import col

model = LogisticRegression().fit(train_data)
preds = model.transform(test_data)
# 提取向量索引为1的数值（正例概率），转为Double类型
preds = preds.withColumn("probability", col("probability").getItem(1).cast("double"))

all_preds = empty.unionAll(preds)

你当前每次迭代union的写法会导致Spark的血缘依赖链越来越长，迭代次数多了之后性能会大幅下降，推荐用以下方式优化：

• 每次循环生成预测结果后，先写入临时分区存储（比如parquet格式），所有迭代完成后统一读取所有临时文件合并为一个DataFrame
• 如果数据量不大，可以把每次生成的preds放到Python列表里，所有迭代完成后一次性合并，示例代码：

preds_list = []
for 每次迭代的业务逻辑:
    model = LogisticRegression().fit(train_data)
    preds = model.transform(test_data)
    # 按需做字段类型转换后加入列表
    preds_list.append(preds)

# 一次性合并所有预测结果，用unionByName可避免列顺序错位问题
all_preds = preds_list[0].unionByName(*preds_list[1:])

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