向 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()))schemaStructType(List(StructField(foo,IntegerType,true)))如您所见，该schema对象已被修改。add您应该重建架构而不是使用方法：schema_new = StructType(schema.fields + [StructField("id_col", LongType(), False)])或者，您可以创建对象的深层副本：import copyold_schema = StructType()new_schehma = copy.deepcopy(old_schema).add(StructField("foo", IntegerType()))old_schemaStructType(List())new_schehmaStructType(List(StructField(foo,IntegerType,true)))