pyspark 将事务转换为列表列表答案

【问题标题】：pyspark convert transactions into a list of listpyspark 将事务转换为列表列表
【发布时间】：2023-03-06 13:02:02
【问题描述】：

我想在 pyspark 中使用 PrefixSpan 序列挖掘。我需要的数据格式如下：

[[['a', 'b'], ['c']],  [['a'], ['c', 'b'], ['a', 'b']],  [['a', 'b'], ['e']], [['f']]]

其中最里面的元素是productId，然后是订单（包含产品列表），然后是客户（包含订单列表）。

我的数据有交易格式：

clientId orderId product

其中 orderId 有多行用于单独的产品，clientId 有多行用于单独的订单。

样本数据：

test = sc.parallelize([[u'1', u'100', u'a'],
 [u'1', u'100', u'a'],
 [u'1', u'101', u'b'],
 [u'2', u'102', u'c'],
 [u'3', u'103', u'b'],
 [u'3', u'103', u'c'],
 [u'4', u'104', u'a'],
 [u'4', u'105', u'b']]
)

到目前为止我的解决方案：
1. 按订单对产品进行分组：

order_prod = test.map(lambda x: [x[1],([x[2]])])
order_prod = order_prod.reduceByKey(lambda a,b: a + b)
order_prod.collect()

导致：

[(u'102', [u'c']),
 (u'103', [u'b', u'c']),
 (u'100', [u'a', u'a']),
 (u'104', [u'a']),
 (u'101', [u'b']),
 (u'105', [u'b'])]

2. 客户群下单：

client_order = test.map(lambda x: [x[0],[(x[1])]])
df_co = sqlContext.createDataFrame(client_order)
df_co = df_co.distinct()
client_order = df_co.rdd.map(list)
client_order = client_order.reduceByKey(lambda a,b: a + b)
client_order.collect()

导致：

[(u'4', [u'105', u'104']),
 (u'3', [u'103']),
 (u'2', [u'102']),
 (u'1', [u'100', u'101'])]

那我想要一个这样的列表：

[[[u'a', u'a'],[u'b']], [[u'c']], [[u'b', u'c']], [[u'a'],[u'b']]]

【问题讨论】：

标签： apache-spark lambda pyspark sequence reduce

【解决方案1】：

这是使用 PySpark 数据框的解决方案（不是我使用 PySpark 2.1）。首先，您必须将 RDD 转换为 Dataframe。

df = test.toDF(['clientId', 'orderId', 'product'])

这是对数据帧进行分组的 sn-p。基本思想是首先按clientId 和orderId 分组，然后将product 列聚合在一起。然后仅按clientId 再次分组。

import pyspark.sql.functions as func

df_group = df.groupby(['clientId', 'orderId']).agg(func.collect_list('product').alias('product_list'))
df_group_2 = df_group[['clientId', 'product_list']].\
    groupby('clientId').\
    agg(func.collect_list('product_list').alias('product_list_group')).\
    sort('clientId', ascending=True)
df_group_2.rdd.map(lambda x: x.product_list_group).collect() # collect output here

结果如下：

[[['a', 'a'], ['b']], [['c']], [['b', 'c']], [['b'], ['a']]]

【讨论】：