CODEWITHSUNDEEP
apache-sparkapache-spark-sqlpyspark
Kyle.
Kyle.

Reputation: 156

Sparse by column to dense array in pyspark

I have two data frames that I need to get information from to generate a third. The first data frame contains information on item iteractions by user, e.g., 

+-----+-----------+-----------+
|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 |
+-----+-----------+-----------+

It has no particular order, and each user has multiple rows. The second data frame is just a list of items with an index, e.g.,

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

This dataframe is quite long, and not every item is represented in the item interaction data frame. What is need is a third data frame where each row contains a vectorized representation of a user's item interactions as an array within a single column, e.g., 

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

Where the array has a 1 if the user interacted with the item at that index, otherwise 0. Is there an easy way to do this in pyspark?

Upvotes: 1

Views: 1124

Answers (3)

OO7
OO7

Reputation: 690

Try this one! You can also modify or make any correction if needed. 

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

Enjoy and cheers!

Update: Set Spark Configuration:

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

Performance Tuning

Upvotes: 1

jxc
jxc

Reputation: 13998

IIUC, you can use pyspark.ml.feature.CountVectorizer to help create the desired vector. Assume df1 is the first dataframe (user, itemId and date) and df2 the 2nd dataframe(itemId and 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])      |
+----+------------------------------+-------------------------+

This creates a SparseVector, if you want an ArrayType column, you will need an 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]|
+----+------------------------------+-------------------------+------------+

Upvotes: 1

blackbishop
blackbishop

Reputation: 32660

You could join the 2 DataFrames and then collect list of indexes group by 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"))

Now using the array of indexes to create new array interactions like this:

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

Upvotes: 0

Related Questions