Pyspark：访问 UDF 中行内的列

Question

pyspark 的初学者试图理解 UDF：

我有一个 PySpark 数据框 p_b，我通过传递数据框的所有行来调用 UDF。我想从行访问列debit。出于某种原因，这没有发生。请在下面找到 sn-ps。

p_b has 4 columns, id, credit, debit,sum

功能：

def test(row): return('123'+row['debit'])

转换为 UDF

test_udf=udf(test,IntegerType())

在数据帧 p_b 上调用 UDF

vals=test_udf(struct([p_b[x] for x in p_b.columns])) print(type(vals)) print(vals)

输出

Column<b'test(named_struct(id, credit,debit,sum))'>

Answer 1

让我们先制作一个示例数据框：

from pyspark.sql.functions import *
from pyspark.sql.types import *  
schema = StructType([StructField("id", StringType(), True),\
                               StructField("credit", IntegerType(), True),\
                       StructField("debit", IntegerType(), True),\
                     StructField("sum", IntegerType(), True)])
df = spark.createDataFrame([("user_10",100, 10,110),("user_11",200, 20,220),("user_12",300, 30,330) ], schema)
df.show()

导致：

+-------+------+-----+---+
|     id|credit|debit|sum|
+-------+------+-----+---+
|user_10|   100|   10|110|
|user_11|   200|   20|220|
|user_12|   300|   30|330|
+-------+------+-----+---+

现在，让我们定义将 123 添加到传递给它的值的 udf：

def test(x):
    return(123+x)
test_udf=udf(test,IntegerType())

让我们看看如何使用 UDF：

df2 = df.withColumn( 'debit' , test_udf(col('debit')) )
df2.show()

导致：

+-------+------+-----+---+
|     id|credit|debit|sum|
+-------+------+-----+---+
|user_10|   100|  133|110|
|user_11|   200|  143|220|
|user_12|   300|  153|330|
+-------+------+-----+---+

请注意，现在您可能需要重新计算“sum”列：

df2 = df2.withColumn( 'sum' ,  col('debit')+col('credit') )
df2.show()

导致：

+-------+------+-----+---+
|     id|credit|debit|sum|
+-------+------+-----+---+
|user_10|   100|  133|233|
|user_11|   200|  143|343|
|user_12|   300|  153|453|
+-------+------+-----+---+