Let assume I have a pyspark dataframe (df1) with information of some users as:
+--------+--------+--------+--------+
|user_id |event_id|code |City |
+--------+--------+--------+--------+
| user1| event1 | ABC | LA |
| user1| event2 | ABC | NYC |
| user2| event3 | DEF | LA |
| user2| event4 | GHK | LA |
| user3| event5 | DEF | NYC |
| user3| event6 | DEF | NYC |
| user3| event7 | ABC | LA |
+--------+--------+--------+--------+
In this dataframe we have repetetive user_ids, but event_ids are unique across dataset. Additionally the code and city for each user can be the same or different. I also have another pyspark dataframe (df2) like this based on above table:
+----------+----------+------------+
|event_id1 |event_id2 | user_match |
+----------+----------+------------+
| event1 | event2 | Ture |
| event1 | event4 | False |
| event2 | event3 | False |
| event2 | event7 | False |
| event5 | event6 | True |
| event6 | event1 | False |
+----------+----------+------------+
As you can see, I don't have all combinations. The goal is feature extraction (to detect users) based on their code and city in this way:
+----------+----------+------------+--------+--------+
|event_id1 |event_id2 | user_match |code |City |
+----------+----------+------------+--------+--------+
| event1 | event2 | Ture | Ture | False |
| event1 | event4 | False | False | Ture |
| event2 | event3 | False | False | False |
| event2 | event7 | False | Ture | False |
| event5 | event6 | True | Ture | Ture |
| event6 | event1 | False | False | False |
+----------+----------+------------+--------+--------+
I implemented this using Pandas in PySpark. But I wonder how I can write it using just PySpark APIs:
%spark2.pyspark
# select all or part of train pairs
num_train_samples = pdf2.shape[0]
feats_train_array = pdf2[0:num_train_samples]
# define a temp array
feats = np.zeros((num_train_samples, 1))
# list of feats
#
feats_titles = ["code", "City"]
# extract features
#
for ft in feats_titles:
fvar = ft
for i in range(num_train_samples):
# read rows related to pairs
info_pair0 = pdf1.loc[pdf1['eventId'] == pdf2[i][0]]
info_pair1 = pdf1.loc[pdf1['eventId'] == pdf2[i][1]]
# compare values
feats_pair0 = (info_pair0[fvar].reset_index(drop=True)).iloc[0]
feats_pair1 = (info_pair1[fvar].reset_index(drop=True)).iloc[0]
if (feats_pair0==feats_pair1):
feats[i] = 1
else:
feats[i] = 0
feats_train_array = np.append(feats_train_array, feats, axis=1)
I think this will be a simpler code using PySpark APIs, but I can't figure it out.
Well, I don't know this is simpler but you can do this.
from pyspark.sql.functions import *
df1 = spark.read.option("header","true").option("inferSchema","true").csv("test1.csv")
df2 = spark.read.option("header","true").option("inferSchema","true").csv("test2.csv") \
.withColumn('user_match', col('user_match').cast('boolean'))
df2.join(df1.withColumnRenamed('event_id', 'event_id1').drop('user_id').alias('a'), ['event_id1'], 'inner') \
.join(df1.withColumnRenamed('event_id', 'event_id2').drop('user_id').alias('b'), ['event_id2'], 'inner') \
.withColumn('code_match', when(expr('a.code = b.code'), True).otherwise(False)) \
.withColumn('city_match', when(expr('a.City = b.City'), True).otherwise(False)) \
.select(*df2.columns, 'code_match', 'city_match').show()
+---------+---------+----------+----------+----------+
|event_id1|event_id2|user_match|code_match|city_match|
+---------+---------+----------+----------+----------+
| event1| event2| true| true| false|
| event1| event4| false| false| true|
| event2| event3| false| false| false|
| event2| event7| false| true| false|
| event5| event6| true| true| true|
| event6| event1| false| false| false|
+---------+---------+----------+----------+----------+