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

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），它被附加到dataframe中，它基本上就是每一行的索引号，它最终将id_col移到最左边。在

我使用的数据

^{pr2}$

函数的输出

^{3}$

所有这些都可以正常工作，但问题是当我运行以下两个命令时

>>> 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)一行时，它没有将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)

如果选中the source，您将看到add方法在适当的地方修改了数据。在

简单的例子更容易看出：

^{pr2}$

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)))