如何根据多列对数据进行分组并构造一个新列 - Pyspark

Question

我正在学习 Pyspark，我用它来将 csv 文件读入数据框（带有列名）：

>>> example_df.show(n=5)
+---------------------+------+-------------+---------+---------+-------------+------------+
|                   id| price|       street| locality|town_city|     district|     country|
+---------------------+------+-------------+---------+---------+-------------+------------+
|                bbbb1|295000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|
|                aaaa2|450000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
|                bbbb2|280000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|
|                aaaa3|425000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
|                cccc1| 96000|CHANCERY LANE| HOLBEACH| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
+---------------------+------+-------------+---------+---------+-------------+------------+
only showing top 5 rows

对于相同的位置（=street|locality|town_city|district|country 的相同值），我想创建一个新的location_id，并将相应的字段分组，如下所示：

+-----------+---------------------+------+-------------+---------+---------+-------------+------------+
|location_id|                   id| price|       street| locality|town_city|     district|     country|
+-----------+---------------------+------+-------------+---------+---------+-------------+------------+
|          0|                aaaa2|450000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
|          0|                aaaa3|425000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
|          1|                bbbb1|295000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|
|          1|                bbbb2|280000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|
|          2|                cccc1| 96000|CHANCERY LANE| HOLBEACH| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|
+-----------+---------------------+------+-------------+---------+---------+-------------+------------+
only showing top 5 rows
...

我知道如何创建新列：

from pyspark.sql.functions import monotonically_increasing_id
df = df.WithColumn("location_id", monotonically_increasing_id())

这只是使用0,1,2 etc 创建一个新列，但是我如何将同一位置的数据分组并赋予它们唯一的location_id？谢谢。

Answer 1

如果您只需要每个位置的 id，您可以使用 hash：

from pyspark.sql import functions as F

example_df.withColumn('location_id', 
    F.hash('street','locality','town_city','district','country')) \
    .show()

输出：

+-----+------+-------------+---------+---------+-------------+------------+-----------+
|   id| price|       street| locality|town_city|     district|     country|location_id|
+-----+------+-------------+---------+---------+-------------+------------+-----------+
|bbbb1|295000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|  406601501|
|aaaa2|450000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|  -97363104|
|bbbb2|280000|   HAYES YARD|   INGHAM|  LINCOLN| WEST LINDSEY|LINCOLNSHIRE|  406601501|
|aaaa3|425000|   MARKET WAY|PINCHBECK| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE|  -97363104|
|cccc1| 96000|CHANCERY LANE| HOLBEACH| SPALDING|SOUTH HOLLAND|LINCOLNSHIRE| 1396052469|
+-----+------+-------------+---------+---------+-------------+------------+-----------+

编辑：请记住hash 函数可能会导致hash collisions