PySpark:withColumn()有两个条件和三个结果

Question

我正在使用Spark和PySpark.我试图实现相当于以下伪代码的结果:

df = df.withColumn('new_column', 
    IF fruit1 == fruit2 THEN 1, ELSE 0. IF fruit1 IS NULL OR fruit2 IS NULL 3.)

我试图在PySpark中这样做,但我不确定语法.有什么指针吗？我调查expr()但无法让它工作.

请注意,这df是一个pyspark.sql.dataframe.DataFrame.

Answer 1

有几种有效的方法可以实现这一点.让我们从必需的导入开始:

from pyspark.sql.functions import col, expr, when

你可以IF在expr中使用Hive 函数:

new_column_1 = expr(
    """IF(fruit1 IS NULL OR fruit2 IS NULL, 3, IF(fruit1 = fruit2, 1, 0))"""
)

或when+ otherwise:

new_column_2 = when(
    col("fruit1").isNull() | col("fruit2").isNull(), 3
).when(col("fruit1") == col("fruit2"), 1).otherwise(0)

最后你可以使用以下技巧:

from pyspark.sql.functions import coalesce, lit

new_column_3 = coalesce((col("fruit1") == col("fruit2")).cast("int"), lit(3))

使用示例数据:

df = sc.parallelize([
    ("orange", "apple"), ("kiwi", None), (None, "banana"), 
    ("mango", "mango"), (None, None)
]).toDF(["fruit1", "fruit2"])

您可以使用如下:

(df
    .withColumn("new_column_1", new_column_1)
    .withColumn("new_column_2", new_column_2)
    .withColumn("new_column_3", new_column_3))

结果是:

+------+------+------------+------------+------------+
|fruit1|fruit2|new_column_1|new_column_2|new_column_3|
+------+------+------------+------------+------------+
|orange| apple|           0|           0|           0|
|  kiwi|  null|           3|           3|           3|
|  null|banana|           3|           3|           3|
| mango| mango|           1|           1|           1|
|  null|  null|           3|           3|           3|
+------+------+------------+------------+------------+

Answer 2

您将要使用如下的udf

from pyspark.sql.types import IntegerType
from pyspark.sql.functions import udf

def func(fruit1, fruit2):
    if fruit1 == None or fruit2 == None:
        return 3
    if fruit1 == fruit2:
        return 1
    return 0

func_udf = udf(func, IntegerType())
df = df.withColumn('new_column',func_udf(df['fruit1'], df['fruit2']))

Answer 3

在pyspark的withColumn功能可以使一个新的变量，条件，加在的时候和其他方式的功能，你有一个正常工作，如果再else结构。对于所有这些，您需要导入 sparksql 函数，因为您将看到以下代码在没有 col() 函数的情况下将无法工作。在第一位，我们声明一个新列——'new column'，然后给出包含在when函数中的条件（即fruit1==fruit2）然后如果条件为真则给出1，如果不为真则控制转到否则然后使用 isNull() 函数处理第二个条件（fruit1 或fruit2 为 Null），如果返回 true 3，如果返回 false，则再次检查否则返回 0 作为答案。

from pyspark.sql import functions as F
df=df.withColumn('new_column', 
    F.when(F.col('fruit1')==F.col('fruit2'), 1)
    .otherwise(F.when((F.col('fruit1').isNull()) | (F.col('fruit2').isNull()), 3))
    .otherwise(0))