时间戳

首页 > 解决方案 > pyspark 中的计数器以检查具有重复项的数组中的数组

python - pyspark 中的计数器以检查具有重复项的数组中的数组

问题描述

我有一个数据框 df1

id     transactions
1      [1, 3,3,3,2,5]
2      [1,2]
root
 |-- id: int (nullable = true)
 |-- transactions: array (nullable = false)
      |-- element: string(containsNull = true)
None

我有一个数据框 df2

items         cost
[1, 3,3, 5]    2
[1, 5]      1

root
|-- items: array (nullable = false)
  |-- element: string (containsNull = true)
 |-- cost: int (nullable = true)
None

我必须检查这些物品是否在交易中，如果是的话，总结成本。[1,3,3,3,5] 中的 [1,3,3,5] 为真，[1,2] 中的 [1,3,3,5] 为假，依此类推。

结果应该是

id     transactions   score
1      [1,3,3,3,5]    3
2      [1,2]          null

我尝试了explode和join（inner，left_semi）方法，但由于重复而全部失败。检查另一个数组 pyspark issubset() 中存在的数组的所有元素， array_intersect() 也不起作用。

我遇到了 Python - 验证一个列表是否是另一个列表的子集。我发现以下解决了问题并且效率很高。

from collections import Counter
not Counter([1,3,3,3,5])-Counter([1,3,3,4,5])
False
>>> not Counter([1,3,3,3,5])-Counter([1,3,3,5])
False
>>> not Counter([1,3,3,5])-Counter([1,3,3,3,5])
True

我尝试了以下

@udf("boolean")
def contains_all(x, y):
if x is not None and y is not None:
    return not (lambda y: dict(Counter(y)))-(lambda x: dict(Counter(x)))


(df1
.crossJoin(df2).groupBy("id", "transactions")
.agg(sum_(when(
    contains_all("transactions", "items"), col("cost")
)).alias("score"))
.show())

但它会引发错误。文件“”，第 39 行，在 contains_all 类型错误中：不支持的操作数类型用于 -: 'function' 和 'function'

任何其他方式来实现这一目标？

标签： pythonsqlcollectionspysparkcounter

解决方案

刚刚更新了 udf 以保留重复项并且不确定性能，

from pyspark.sql.functions import udf,array_sort,sum as sum_,when,col

dff = df1.crossjoin(df2)

dff = dff.withColumn('transaction',array_sort('transaction')).\
      withColumn('items',array_sort('items')) ## sorting here,it's needed in UDF

+---+---------------+------------+----+
| id|    transaction|       items|cost|
+---+---------------+------------+----+
|  1|[1, 2, 3, 3, 5]|[1, 3, 3, 5]|   2|
|  1|[1, 2, 3, 3, 5]|      [1, 5]|   1|
|  2|         [1, 2]|[1, 3, 3, 5]|   2|
|  2|         [1, 2]|      [1, 5]|   1|
+---+---------------+------------+----+

@udf('boolean')
def is_subset_w_dup(trans,itm):
    itertrans = iter(trans)
    return all(i in itertrans for i in itm)


dff.groupby('id','transaction').agg(sum_(when(is_subset_w_dup('transaction','items'),col('cost'))).alias('score')).show()

+---+---------------+-----+
| id|    transaction|score|
+---+---------------+-----+
|  2|         [1, 2]| null|
|  1|[1, 2, 3, 3, 5]|    3|
+---+---------------+-----+

推荐阅读