PySpark：如何在 PySpark SQL 中创建计算列？

Question

使用 PySpark SQL 并给定 3 列，我想创建一个额外的列来划分两列，第三列是 ID 列。

df = sqlCtx.createDataFrame(
    [
        (1, 4, 2),
        (2, 5, 2),
        (3, 10, 4),
        (4, 50, 10)
    ],
    ('ID', 'X', 'Y')
)

这是所需的输出：

+----+----+----+---------------------+
| ID | x  | y  | z (expected result) |
+----+----+----+---------------------+
|  1 |  4 |  2 | 2                   |
|  2 |  5 |  2 | 2.5                 |
|  3 | 10 |  4 | 2.5                 |
|  4 | 50 | 10 | 5                   |
+----+----+----+---------------------+

为此，我创建了一个 UDF：

def createDivision(args):
    X = float(args[0])
    Y = float(args[1])
    RESULT = X / Y
    return RESULT

udf_createDivision = udf(createDivision, FloatType())

udf_createDivision_calc = udf_createDivision(df['X'], df['Y'])

df = df.withColumn("Z", udf_createDivision_calc)

df.show()

然后我在输出中得到一个长错误：

Py4JJavaError: An error occurred while calling o7401.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 756.0 failed 1 times, most recent failure: Lost task 0.0 in stage 756.0 (TID 7249, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/opt/spark/spark-2.4.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 372, in main
    process()
  File "/opt/spark/spark-2.4.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 367, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/opt/spark/spark-2.4.0-bin-hadoop2.7/python/lib/pyspark.zip/pyspark/worker.py", line 243, in <lambda>
    func = lambda _, it: map(mapper, it)
  File "<string>", line 1, in <lambda>.......

我非常感谢一些帮助，因为我不知道如何解释错误。谢谢。

Answer 1

只是表达：

from pyspark.sql.functions import col

df.withColumn("Z", col("x") / col("y"))

至于你的代码（你真的不应该在这里使用 udf ）它应该是：

def createDivision(x, y):
    return x / y

或者

def createDivision(*args):
    return args[0] / args[1]