PySpark - 获取具有相同值的数组元素的数量

Posted 2023-04-15

【中文标题】PySpark - 获取具有相同值的数组元素的数量

【英文标题】：PySpark - getting number of elements of array with same value

【发布时间】：2020-07-29 15:23:41

【问题描述】：

我正在学习 Spark，但遇到了无法克服的问题。 我想要实现的是在相同位置为 2 个数组获取具有相同值的元素数量。我可以通过 Python UDF 获得我想要的东西，但我想知道是否有一种方法只使用 Spark 函数。

df_bits = sqlContext.createDataFrame([[[0, 1, 1, 0, 0, ],
                                       [1, 1, 1, 0, 1, ],
                                     ]],['bits1', 'bits2'])
df_bits_with_result = df_bits.select('bits1', 'bits2', some_magic('bits1', 'bits2').show()


+--------------------+--------------------+---------------------------------+
|bits1                  |bits2                  |some_magic(bits1, bits2)|
+--------------------+--------------------+---------------------------------+
|[0, 1, 1, 0, 1, ]    |[1, 1, 1, 0, 0, ]   |3                                      |
+--------------------+--------------------+---------------------------------+

为什么是 3？ bits1[1] == bits2[1] AND bits1[2] == bits2[2] AND bits1[3] == bits2[3] 我尝试玩 rdd.reduce 但没有运气。

【问题讨论】：

什么版本的火花？对于通用解决方案，您可以在 2.4+ 中使用 arrays_zip。也请format your code properly.

我会补充一点，如果您的问题像您在这种情况下所示的那样简化，其中 a) 数组内容都是非空二进制值（即 1 和 0）和 b)数组是相同的、固定的、已知的长度，你也可以做一些 hacky 正则表达式或循环

我忘了提及我使用的是哪个版本的 Spark，但幸运的是我使用的是 2.4+，所以当前的两个答案都适用于我。

【参考方案1】：

也许这有帮助-

spark>=2.4

使用 aggregate 和 zip_with

 val df = spark.sql("select array(0, 1, 1, 0, 0, null) as bits1, array(1, 1, 1, 0, 1, null) as bits2")
    df.show(false)
    df.printSchema()

    /**
      * +----------------+----------------+
      * |bits1           |bits2           |
      * +----------------+----------------+
      * |[0, 1, 1, 0, 0,]|[1, 1, 1, 0, 1,]|
      * +----------------+----------------+
      *
      * root
      * |-- bits1: array (nullable = false)
      * |    |-- element: integer (containsNull = true)
      * |-- bits2: array (nullable = false)
      * |    |-- element: integer (containsNull = true)
      */

    df.withColumn("x", expr("aggregate(zip_with(bits1, bits2, (x, y) -> if(x=y, 1, 0)), 0, (acc, x) -> acc + x)"))
      .show(false)

    /**
      * +----------------+----------------+---+
      * |bits1           |bits2           |x  |
      * +----------------+----------------+---+
      * |[0, 1, 1, 0, 0,]|[1, 1, 1, 0, 1,]|3  |
      * +----------------+----------------+---+
      */

【参考方案2】：

Pyspark 使用 arrays_zip 如 cmets 中所述

from pyspark.sql import functions as F

df_bits.withColumn("sum", \
              F.expr("""aggregate(arrays_zip(bits1,bits2),0,(acc,x)-> IF(x.bits1==x.bits2,1,0)+acc)""")).show()

#+---------------+---------------+---+
#|          bits1|          bits2|sum|
#+---------------+---------------+---+
#|[0, 1, 1, 0, 0]|[1, 1, 1, 0, 1]|  3|
#+---------------+---------------+---+