python - PySpark:在 DataFrame 中的小组内部进行迭代

Question

我试图了解如何在 PySpark DataFrame 中的小组内进行操作。假设我有具有以下架构的 DF:

root
|-- first_id: string (nullable = true)
|-- second_id_struct: struct (nullable = true)
|    |-- s_id: string (nullable = true)
|    |-- s_id_2: int (nullable = true)
|-- depth_from: float (nullable = true)
|-- depth_to: float (nullable = true)
|-- total_depth: float (nullable = true) 

所以数据可能看起来像这样:

我愿意:

1. 按first_id对数据进行分组
2. 在每个组内，按 s_id_2 升序排列
3. 将额外的列layer附加到结构或根DataFrame，以指示该s_id_2在组中的顺序。

例如:

first_id | second_id | second_id_order 
---------| --------- | ---------------
      A1 |   [B, 10] | 1  
---------| --------- | ---------------
      A1 |   [B, 14] | 2
---------| --------- | ---------------
      A1 |   [B, 22] | 3
---------| --------- | ---------------
      A5 |    [A, 1] | 1
---------| --------- | ---------------
      A5 |    [A, 7] | 2
---------| --------- | ---------------
      A7 |      null | 1
---------| --------- | ---------------        

分组后，每个 first_id 将最多有 4 个 second_id_struct。我该如何解决这类问题？

我对如何在 DataFrame 的小组(1-40 行)内进行迭代操作特别感兴趣，其中组内的列顺序很重要。

谢谢!

Answer 1

创建一个数据框

d = [{'first_id': 'A1', 'second_id': ['B',10]}, {'first_id': 'A1', 'second_id': ['B',14]},{'first_id': 'A1', 'second_id': ['B',22]},{'first_id': 'A5', 'second_id': ['A',1]},{'first_id': 'A5', 'second_id': ['A',7]}]

df = sqlContext.createDataFrame(d) 

你可以看到结构

df.printSchema()

|-- first_id: string (nullable = true)
|-- second_id: array (nullable = true)
|........|-- element: string (containsNull = true)

df.show()
+--------+----------+
|first_id|second_id |
+--------+----------+
|      A1|   [B, 10]|
|      A1|   [B, 14]|
|      A1|   [B, 22]|
|      A5|    [A, 1]|
|      A5|    [A, 7]|
+--------+----------+

然后您可以使用dense_rank和Window函数来显示子组中的顺序。与SQL中的over分区相同。

窗函数介绍:Introducing Window Functions in Spark SQL

代码在这里:

# setting a window spec
windowSpec = Window.partitionBy('first_id').orderBy(df.second_id[1])
# apply dense_rank to the window spec
df.select(df.first_id, df.second_id, dense_rank().over(windowSpec).alias("second_id_order")).show()

结果:

+--------+---------+---------------+
|first_id|second_id|second_id_order|
+--------+---------+---------------+
|      A1|  [B, 10]|              1|
|      A1|  [B, 14]|              2|
|      A1|  [B, 22]|              3|
|      A5|   [A, 1]|              1|
|      A5|   [A, 7]|              2|
+--------+---------+---------------+