按列稀疏到pyspark中的密集数组

Question

我有两个数据框，我需要从中获取信息以生成第三个数据框。第一个数据帧包含有关用户项目迭代的信息，例如

+-----+-----------+-----------+
|user | itemId    |date       |
+-----+-----------+-----------+
|1    | 10005880  |2019-07-23 |
|2    | 10005903  |2019-07-23 |
|3    | 10005903  |2019-07-23 |
|1    | 12458442  |2019-07-23 |
|1    | 10005903  |2019-07-26 |
|3    | 12632813  |2019-07-26 |
|2    | 12632813  |2019-07-26 |
+-----+-----------+-----------+

它没有特定的顺序，每个用户都有多行。第二个数据框只是带有索引的项目列表，例如

+-----------+-----------+
| itemId    |index      |
+-----------+-----------+
| 10005880  |1          |
| 10005903  |2          |
| 12458442  |3          |
|    ...    |   ...     |
| 12632813  |2000000    |
+-----------+-----------+

此数据帧很长，并且并非每个项目都在项目交互数据框中表示。所需要的是第三数据帧，其中每一行包含用户项交互的矢量化表示，作为单列内的数组，例如

+-----+--------------------+
|user |  interactions      |
+-----+--------------------+
|1    |  <1, 1, 1, ..., 0> |                        
|2    |  <0, 1, 0, ..., 1> |                         
|3    |  <0, 1, 0, ..., 1> |                            
+-----+--------------------+

如果用户与该索引处的项目进行交互，则数组的位置为1，否则为0。在pyspark中有一种简单的方法吗？

Answer 1

IIUC，您可以使用pyspark.ml.feature。CountVectorizer帮助创建所需的向量。假设df1是第一个数据帧（用户，itemId和日期），而df2是第二个数据帧（itemId和索引）：

from pyspark.ml.feature import CountVectorizerModel
from pyspark.sql.functions import collect_set

df3 = df1.groupby('user').agg(collect_set('itemId').alias('items_arr'))

# set up the vocabulary from the 2nd dataframe and then create CountVectorizerModel from this list
# set binary=True so that this is doing the same as OneHotEncoder
voc = [ r.itemId for r in df2.select('itemId').sort('index').collect() ]
model = CountVectorizerModel.from_vocabulary(voc, inputCol='items_arr', outputCol='items_vec', binary=True)

df_new = model.transform(df3)
df_new.show(truncate=False)
+----+------------------------------+-------------------------+
|user|items_arr                     |items_vec                |
+----+------------------------------+-------------------------+
|3   |[10005903, 12632813]          |(4,[1,2],[1.0,1.0])      |
|1   |[10005903, 12458442, 10005880]|(4,[0,1,3],[1.0,1.0,1.0])|
|2   |[10005903, 12632813]          |(4,[1,2],[1.0,1.0])      |
+----+------------------------------+-------------------------+

这将创建一个SparseVector，如果需要ArrayType列，则需要一个udf：

from pyspark.sql.functions import udf
udf_to_array = udf(lambda v: [*map(int, v.toArray())], 'array<int>')

df_new.withColumn('interactions', udf_to_array('items_vec')).show(truncate=False)
+----+------------------------------+-------------------------+------------+
|user|items_arr                     |items_vec                |interactions|
+----+------------------------------+-------------------------+------------+
|3   |[10005903, 12632813]          |(4,[1,2],[1.0,1.0])      |[0, 1, 1, 0]|
|1   |[10005903, 12458442, 10005880]|(4,[0,1,3],[1.0,1.0,1.0])|[1, 1, 0, 1]|
|2   |[10005903, 12632813]          |(4,[1,2],[1.0,1.0])      |[0, 1, 1, 0]|
+----+------------------------------+-------------------------+------------+

Answer 2

尝试这个！如果需要，您还可以修改或进行任何更正。

from pyspark.sql.functions import col, when, arrays_zip

userIndexes = users.join(items, users.itemId == items.itemId, 'left').crosstab('user', 'index')

cols = userIndexes.columns.filter(_ != 'user')

userIndexes.select('user', arrays_zip([when(col(c).isNull(), lit(0)).otherwise(lit(1)) for c in cols]).alias('interactions')).show()

享受欢呼！

更新：设置Spark配置：

var sparkConf: SparkConf = null
sparkConf = new SparkConf()
.set("spark.sql.inMemoryColumnarStorage.batchSize", 36000)

Performance Tuning

Answer 3

您可以加入2个数据框，然后按user收集索引组列表。

df_users_items = df_users.join(df_items, ["itemId"], "left")

df_user_interations = df_users_items.groupBy("user").agg(collect_set("index").alias("interactions"))

现在使用索引数组创建新数组interactions，如下所示：

max_index = df_items.select(max(col("index")).alias("max_index")).first().max_index
interactions_col = array(
    *[when(array_contains("interactions", i + 1), lit(1)).otherwise(lit(0)) for i in range(max_index)])

df_user_interations.withColumn("interactions", interactions_col).show()