向 StructType 添加新列时的行为不明确

Question

我在 PySpark 中定义了一个函数，它是-def add_ids(X):    schema_new = X.schema.add("id_col", LongType(), False)    _X = X.rdd.zipWithIndex().map(lambda l: list(l[0]) + [l[1]]).toDF(schema_new)    cols_arranged = [_X.columns[-1]] + _X.columns[0:len(_X.columns) - 1]    return _X.select(*cols_arranged)在上面的函数中，我正在创建一个新列（名称为id_col），该列附加到数据框，它基本上只是每行的索引号，最后将 移到id_col最左侧。我正在使用的数据>>> X.show(4)+-----------+-------+-------------+-------------+-------+----+------------------------+---+-------+|Pregnancies|Glucose|BloodPressure|SkinThickness|Insulin| BMI|DiabetesPedigreeFunction|Age|Outcome|+-----------+-------+-------------+-------------+-------+----+------------------------+---+-------+|          6|    148|           72|           35|      0|33.6|                   0.627| 50|      1||          1|     85|           66|           29|      0|26.6|                   0.351| 31|      0||          8|    183|           64|            0|      0|23.3|                   0.672| 32|      1||          1|     89|           66|           23|     94|28.1|                   0.167| 21|      0|+-----------+-------+-------------+-------------+-------+----+------------------------+---+-------+only showing top 4 rows所有这些工作正常，但问题是当我运行以下两个命令时>>> X.columns['Pregnancies', 'Glucose', 'BloodPressure', 'SkinThickness', 'Insulin', 'BMI', 'DiabetesPedigreeFunction', 'Age', 'Outcome', 'id_col']如果您查看 的结果X.columns，您会id_col在最后注意到。但是当我之前运行X.show(4)a 行时，它没有显示id_col为一列。现在，当我尝试运行时add_ids(X).show(4)，出现以下错误pyspark.sql.utils.AnalysisException: "Reference 'id_col' is ambiguous, could be: id_col, id_col.;"我做错了什么？

Answer 1

错误在这里：

schema_new = X.schema.add("id_col", LongType(), False)

如果您检查源，您将看到该add方法修改了适当的数据。

在简化的示例中更容易看到：

from pyspark.sql.types import *

schema = StructType()

schema.add(StructField("foo", IntegerType()))

schema

StructType(List(StructField(foo,IntegerType,true)))

如您所见，该schema对象已被修改。

add您应该重建架构而不是使用方法：

schema_new = StructType(schema.fields + [StructField("id_col", LongType(), False)])

或者，您可以创建对象的深层副本：

import copy

old_schema = StructType()

new_schehma = copy.deepcopy(old_schema).add(StructField("foo", IntegerType()))

old_schema

StructType(List())

new_schehma

StructType(List(StructField(foo,IntegerType,true)))