从 Spark / PySpark 中不一致数组的数据帧列中比较和删除元素答案

【问题标题】：Compare and remove elements out of dataframe columns of inconsistent arrays in Spark / PySpark从 Spark / PySpark 中不一致数组的数据帧列中比较和删除元素
【发布时间】：2020-01-30 09:05:22
【问题描述】：

我是 Spark 的新手，我找不到解决问题的方法，非常感谢任何建议或帮助。

我有一个 Pyspark.sql.dataframe，其中包含两个数组列，其中包含字符串。两个列数组的长度不一致，某些行也有 Null 条目。我需要比较这两列，并且必须为 B 列中的每一行删除数组的一个元素，当它在 OVERRIDE 列的数组中的该行中找到时。

+---------------+---------------+
|    OVERRIDE   |         B     |
+---------------+---------------+
|          ['a']|      ['a','b']|
|           null|          ['b']|
|           null|      ['a','c']|
|      ['d','g']|      ['d','g']|
|           null|           null|
|          ['f']|          ['f']|
+---------------+---------------+

最后应该是这样的：

+---------------+---------------+
|    OVERRIDE   |         B     |
+---------------+---------------+
|          ['a']|          ['b']|
|           null|          ['b']|
|           null|      ['a','c']|
|      ['d','g']|           null|
|           null|           null|
|          ['f']|           null|
+---------------+---------------+

我试过了

from pyspark.sql.functions import array_remove, array_intersect

df = df.withColumn('B', array_remove(df.B, df.OVERRIDE))

还有

df = df.withColumn('B', array_remove(df.B, array_intersect(df.OVERRIDE, df.B)))

但了解到 array_remove() 不能遍历列，而是只能取一个元素（例如“a”）将其删除，然后在 B 列的所有行中。

我是否必须构建一个 udf 函数，如果是，我应该怎么做？

【问题讨论】：

标签： python pyspark apache-spark-sql pyspark-sql higher-order-functions

【解决方案1】：

您可以使用udf 来执行此操作

@udf(returnType=ArrayType(StringType()))
def removeFromRight(override,b):
if(override==None or b==None):
    return b

filtered_list=[x for x in b if x not in override]
if(len(filtered_list)==0):
    filtered_list=None
return filtered_list

test1=test.withColumn("new_overridden_col",removeFromRight(col("override"),col("b")))    
test1.show()

//output of test1
+--------+------+------------------+
|override|     b|new_overridden_col|
+--------+------+------------------+
|     [a]|[a, b]|               [b]|
|    null|   [b]|               [b]|
|    null|[a, c]|            [a, c]|
|  [d, g]|  null|              null|
|    null|  null|              null|
|     [f]|  null|              null|
+--------+------+------------------+

【讨论】：

非常感谢！奇怪的是，在添加第 3 行和第 4 行之前，它适用于一个小样本集，但不适用于完整数据集。每次我想使用新列“new_overridden_col”访问/显示或计算时，我都会遇到一堆错误，但只有那时。

【解决方案2】：

如果您使用的是 Spark >= 2.4.0，则可以使用内置 array_except(a, b)。该函数将返回 a 中存在但 b 中不存在的所有项目。虽然该函数仅在两个参数都没有空值时才起作用，因此在使用它之前我们需要将 null 替换为空数组。

这里是python版本：

from pyspark.sql.functions import array_except, when, array, col

df = spark.createDataFrame([
  [["a"], ["a", "b"]],
 [None, ["b"]],
 [None, ["a", "c"]],
 [["d", "g"], ["d", "g"]],
 [["f"], ["f"]]
], ["OVERRIDE", "B"])

df.withColumn("OVERRIDE", when(col("OVERRIDE").isNull(), array()).otherwise(col("OVERRIDE"))) \
  .withColumn("diff", array_except(col("B"), col("OVERRIDE"))) \
  .show()

// +--------+------+------+
// |OVERRIDE|     B|  diff|
// +--------+------+------+
// |     [a]|[a, b]|   [b]|
// |      []|   [b]|   [b]|
// |      []|[a, c]|[a, c]|
// |  [d, g]|[d, g]|    []|
// |     [f]|   [f]|    []|
// +--------+------+------+

还有 Scala 的：

import org.apache.spark.sql.functions.{array_except, when, array}

val df = Seq(
 (Seq("a"), Seq("a", "b")),
 (null, Seq("b")),
 (null, Seq("a", "c")),
 (Seq("d", "g"), Seq("d", "g")),
 (Seq("f"), Seq("f"))
).toDF("OVERRIDE", "B")

df.withColumn("OVERRIDE", when($"OVERRIDE".isNull, array()).otherwise($"OVERRIDE"))
  .withColumn("diff", array_except($"B", $"OVERRIDE"))
  .show

// +--------+------+------+
// |OVERRIDE|     B|  diff|
// +--------+------+------+
// |     [a]|[a, b]|   [b]|
// |      []|   [b]|   [b]|
// |      []|[a, c]|[a, c]|
// |  [d, g]|[d, g]|    []|
// |     [f]|   [f]|    []|
// +--------+------+------+

【讨论】：