合并多组NumPy数组创建PySpark DataFrame时的类型错误求助

核心问题 #

你存在两个关键错误：

1. Schema定义与数组形状不匹配：你嵌套了两层ArrayType，但实际数组大多是一维结构（仅最外层是行维度）。比如target_id是(4621998,)的一维数组，每行对应单个数值而非数组；user_feature每行是长度620的一维数组，而非二维数组。
2. Spark不支持直接读取numpy.ndarray：Spark的ArrayType仅接受Python原生序列（如list），无法直接解析numpy数组对象，必须先转换为Python列表。

修正步骤 #

1. 重新定义正确的Schema #

根据你的数组形状，正确的Schema如下：

from pyspark.sql.types import StructType, StructField, ArrayType, FloatType

schema_fc = StructType([
    StructField("user_feature", ArrayType(FloatType()), False),
    StructField("sequence_service_id_list", ArrayType(FloatType()), False),
    StructField("target_service_id", FloatType(), False),
    StructField("target_label", FloatType(), False),
    StructField("sequence_label_list", ArrayType(FloatType()), False)
])

2. 将numpy数组转换为Python列表 #

创建DataFrame前，把每个numpy数组转换为Spark可识别的原生类型：

• 二维数组（如user_feature、service_id）：用.tolist()将每行转成列表
• 一维数组（如target_id、target_label）：直接提取单个元素并转为float类型

示例代码：

# 转换numpy数组为Python可迭代对象
user_feature_list = [row.tolist() for row in user_feature]
service_id_list = [row.tolist() for row in service_id]
target_id_list = [float(val) for val in target_id]
target_label_list = [float(val) for val in target_label]
service_label_list = [row.tolist() for row in service_label]

# 创建DataFrame
df = spark.createDataFrame(
    zip(user_feature_list, service_id_list, target_id_list, target_label_list, service_label_list),
    schema_fc
)

验证结果 #

可以通过以下代码确认转换后的DataFrame结构和数据：

df.printSchema()
df.show(5, truncate=False)

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