Разрежьте по столбцу до плотного массива в 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

Попробуйте это! Вы также можете изменить или внести любые исправления, если это необходимо.

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)

Настройка производительности

Answer 2

IIU C, вы можете использовать pyspark.ml.feature. CountVectorizer , чтобы помочь создать нужный вектор. Предположим, что df1 является первым фреймом данных (user, itemId и date), а df2 - вторым фреймом данных (itemId и index):

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 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()