比较基于其他列的 PySpark 数据框的某些列？答案

【问题标题】：Comparing some columns of a PySpark dataframe based on other columns?比较基于其他列的 PySpark 数据框的某些列？
【发布时间】：2020-12-13 19:54:59
【问题描述】：

假设我有一个 pyspark 数据框 (df1)，其中包含一些用户的信息：

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

在这个数据框中，我们有重复的 user_ids，但 event_ids 在数据集中是唯一的。此外，每个用户的代码和城市可以相同或不同。根据上表，我还有另一个像这样的 pyspark 数据框（df2）：

+----------+----------+------------+
|event_id1 |event_id2 | user_match |
+----------+----------+------------+
| event1   | event2   | Ture       |
| event1   | event4   | False      |
| event2   | event3   | False      |
| event2   | event7   | False      |
| event5   | event6   | True       |
| event6   | event1   | False      |
+----------+----------+------------+

如您所见，我没有所有组合。目标是以这种方式基于他们的代码和城市进行特征提取（以检测用户）：

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

我在 PySpark 中使用 Pandas 实现了这一点。但我想知道如何仅使用 PySpark API 编写它：

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

我认为这将是使用 PySpark API 的更简单的代码，但我想不通。

【问题讨论】：

标签： pyspark feature-extraction

【解决方案1】：

嗯，我不知道这更简单，但你可以这样做。

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

【讨论】：