使用用户定义的值在数据框中添加一个新列。 (pyspark)

Question

数组 A1 的三个值来自某个函数 -

A1 = [1,2,3,4]
A1 = [5,6,7,8]
A1 = [1,3,4,1]

我想在其中添加一个包含我的数组值的新列的数据框 -

+---+---+-----+
| x1| x2|   x3|
+---+---+-----+
|  1|  A|  3.0|
|  2|  B|-23.0|
|  3|  C| -4.0|
+---+---+-----+

我这样试过（假设 'df' 是我的数据框）-

for i  in range(0, 2):
   df = df.withColumn("x4", array(lit(A1[0]), lit(A1[1]), lit(A1[2]))

但是这段代码的问题是它正在用数组 'A1' 的最后一个值更新列，就像这样 -

+---+---+-----+---------+
| x1| x2|   x3|       x4|
+---+---+-----+---------+
|  1|  A|  3.0|[1,3,4,1]|
|  2|  B|-23.0|[1,3,4,1]|
|  3|  C| -4.0|[1,3,4,1]|
+---+---+-----+---------+

但我想要这样 -

+---+---+-----+---------+
| x1| x2|   x3|       x4|
+---+---+-----+---------+
|  1|  A|  3.0|[1,2,3,4]|
|  2|  B|-23.0|[5,6,7,8]|
|  3|  C| -4.0|[1,3,4,1]|
+---+---+-----+---------+

我需要在我的代码中添加什么？

Answer 1

查看您的代码，我认为 A1 值至少取决于 x1、x2 或 x3 列中的一列。

因此，您不能使用 A1 定义新列，但可以使用将定义 A1 所需的列作为参数的函数。

这只是一个假设，但也许，您只需要一个字典，A = {1:[1,2,3,4] , 2:[5,6,7,8], 3:[1,3,4,1],}并在您的 withColumn.

的 UDF 中使用它

Answer 2

怎么样：

from pyspark.sql import SparkSession
import pandas as pd

spark = SparkSession.builder.appName('test').getOrCreate()
df=spark.createDataFrame(data=[(1,'A',3),(2,'B',-23),(3,'C',-4)],schema=['x1','x2','x3'])

+---+---+---+
| x1| x2| x3|
+---+---+---+
|  1|  A|  3|
|  2|  B|-23|
|  3|  C| -4|
+---+---+---+

mydict = {1:[1,2,3,4] , 2:[5,6,7,8], 3:[1,3,4,1]}

def addExtraColumn(df,mydict):
    names = df.schema.names
    count=1
    mylst=[]
    for row in df.rdd.collect():
        RW=row.asDict()
        rowLst=[]
        for name in names:
            rowLst.append(RW[name])
        rowLst.append(mydict[count])
        count=count+1
        mylst.append(rowLst)
    return mylst

newlst = addExtraColumn(df,mydict)

df1 = spark.sparkContext.parallelize(newlst).toDF(['x1','x2','x3','x4'])

df1.show()

+---+---+---+------------+
| x1| x2| x3|          x4|
+---+---+---+------------+
|  1|  A|  3|[1, 2, 3, 4]|
|  2|  B|-23|[5, 6, 7, 8]|
|  3|  C| -4|[1, 3, 4, 1]|
+---+---+---+------------+

Answer 3

所以，在我大伤脑筋之后，我发现这不能使用 pyspark 的 withColumn 函数来完成，因为它会创建一个列，但所有列都在同一行。而且我也不能使用 udf 因为我的新列不依赖于现有数据框的任何先前列。

所以我做了这样的事情 - 假设您在 for 循环中获取数组 A1 的不同值（在我的例子中就是这种情况）

f_array = []
for i in range(0,10):
   f_array.extend([(i, A1)])

# Creating a new df for my array.

df1 = spark.createDataFrame(data = f_array, schema = ["id", "x4"])
df1.show()

+---+---------+
| id|       x4|
+---+---------+
|  0|[1,2,3,4]|
|  1|[5,6,7,8]|
|  2|[1,3,4,1]|
+---+---------+
# Suppose no columns matches to our df then creating one extra column named `id` as present in our `df1`. This is used for joining both the dataframes.

df = df.withColumn('id', monotonically_increasing_id())
df.show()

+---+---+---+-----+
| id| x1| x2|   x3|
+---+---+---+-----+
|  0|  1|  A|  3.0|
|  1|  2|  B|-23.0|
|  2|  3|  C| -4.0|
+---+---+---+-----+

# Now join both the dataframes using common column `id`.

df = df.join(df1, df.id == df1.id).drop(df.id).drop(df1.id)
df.show()

+---+---+---+------------+
| x1| x2| x3|          x4|
+---+---+---+------------+
|  1|  A|  3|[1, 2, 3, 4]|
|  2|  B|-23|[5, 6, 7, 8]|
|  3|  C| -4|[1, 3, 4, 1]|
+---+---+---+------------+

使用用户定义的值在数据框中添加一个新列。 (pyspark)

Add a new column in dataframe with user defined values. (pyspark)

python

pyspark

pyspark-sql