Pyspark：多个数组的交集

Question

我有以下测试数据，必须借助pyspark检查以下语句（数据实际上非常大：700000笔交易，每笔交易有10+个产品）：

import pandas as pd
import datetime

data = {'date': ['2014-01-01', '2014-01-02', '2014-01-03', '2014-01-04', '2014-01-05', '2014-01-06'],
     'customerid': [1, 2, 2, 3, 4, 3], 'productids': ['A;B', 'D;E', 'H;X', 'P;Q;G', 'S;T;U', 'C;G']}
data = pd.DataFrame(data)
data['date'] = pd.to_datetime(data['date'])

“某个客户 ID 在 x 天内存在的交易的特点是购物车中至少有一件相同的产品。”

到目前为止，我有以下方法（示例 x = 2）：

spark = SparkSession.builder \
    .master('local[*]') \
    .config("spark.driver.memory", "500g") \
    .appName('my-pandasToSparkDF-app') \
    .getOrCreate()
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
spark.sparkContext.setLogLevel("OFF")

df=spark.createDataFrame(data)

x = 2

win = Window().partitionBy('customerid').orderBy(F.col("date").cast("long")).rangeBetween(-(86400*x), Window.currentRow)
test = df.withColumn("productids", F.array_distinct(F.split("productids", "\;")))\
    .withColumn("flat_col", F.array_distinct(F.flatten((F.collect_list("productids").over(win))))).orderBy(F.col("date"))

test = test.toPandas()

因此，从我们回顾过去 2 天的每笔交易中，按 customerid 分组，相应的产品汇总在“flat_col”列中。

但我真正需要的是具有相同 ID 的购物篮的交集。只有这样我才能判断是否有共同的产品。

所以“flat_col”的第五行应该是['G']而不是['P'，'Q'，'G'，'C']。同样，[] 应该出现在“flat_col”的所有其他行中。

非常感谢！

Answer 1

您可以在没有 self-join 的情况下实现此目的（因为在大数据中连接是昂贵的 shuffle 操作）使用 higher order functions in spark 2.4。使用的函数filter,transform,aggregate。

df=spark.createDataFrame(data)

x = 2

win = Window().partitionBy('customerid').orderBy(F.col("date").cast("long")).rangeBetween(-(86400*x), Window.currentRow)
test = df.withColumn("productids", F.array_distinct(F.split("productids", "\;")))\
    .withColumn("flat_col", F.flatten(F.collect_list("productids").over(win)))\
    .withColumn("occurances", F.expr("""filter(transform(productids, x->\
     IF(aggregate(flat_col, 0,(acc,t)->acc+IF(t=x,1,0))>1,x,null)),y->y!='null')"""))\
    .drop("flat_col").orderBy("date").show()

+-------------------+----------+----------+----------+
|               date|customerid|productids|occurances|
+-------------------+----------+----------+----------+
|2014-01-01 00:00:00|         1|    [A, B]|        []|
|2014-01-02 00:00:00|         2|    [D, E]|        []|
|2014-01-03 00:00:00|         2|    [H, X]|        []|
|2014-01-04 00:00:00|         3| [P, Q, G]|        []|
|2014-01-05 00:00:00|         4| [S, T, U]|        []|
|2014-01-06 00:00:00|         3|    [C, G]|       [G]|
+-------------------+----------+----------+----------+

Answer 2

自加入是有史以来最好的技巧

from pyspark.sql.functions import concat_ws, collect_list
spark.createDataFrame(data).registerTempTable("df")
sql("SELECT date, customerid, explode(split(productids, ';')) productid FROM df").registerTempTable("altered")
df = sql("SELECT al.date, al.customerid, al.productid productids, altr.productid flat_col FROM altered al left join altered altr on altr.customerid = al.customerid and al.productid = altr.productid and al.date != altr.date and datediff(al.date,altr.date) <=2 and datediff(al.date,altr.date) >=-2")
df.groupBy("date", "customerid").agg(concat_ws(",", collect_list("productids")).alias('productids'), concat_ws(",", collect_list("flat_col")).alias('flat_col')).show()