Apache Spark group by DF, собирать значения в список и затем группировать по списку

Question

У меня есть следующий Apache Spark DataFrame ( DF1 ):

function_name | param1 | param2 | param3 | result
---------------------------------------------------
     f1       |   a    |   b    |   c    |   1        
     f1       |   b    |   d    |   m    |   0
     f2       |   a    |   b    |   c    |   0
     f2       |   b    |   d    |   m    |   0
     f3       |   a    |   b    |   c    |   1
     f3       |   b    |   d    |   m    |   1
     f4       |   a    |   b    |   c    |   0
     f4       |   b    |   d    |   m    |   0

Прежде всего, я хотел бы сгруппировать DataFrame по function_name, собрать результаты в ArrayType и получить новый DataFrame ( DF2 ):

function_name | result_list
--------------------------------
     f1       |  [1,0]
     f2       |  [0,0]
     f3       |  [1,1]
     f4       |  [0,0]

, после чего мне нужно собрать function_name в ArrayType, сгруппировав result_list и I Вы получите новый DataFrame, подобный следующему ( DF3 ):

result_list |  function_name_lists
------------------------------------
    [1,0]   |   [f1]
    [0,0]   |   [f2,f4]
    [1,1]   |   [f3]

Итак, у меня есть вопрос - прежде всего, могу ли я использовать группирование по столбцу ArrayType в Apache Spark ? Если это так, я могу потенциально иметь десятки миллионов значений в result_list ArrayType отдельном поле. Сможет ли Apache Spark сгруппировать по столбцу result_list в этом случае?

Answer 1

Да, вы можете сделать это.

Создание вашего фрейма данных:

from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import *
list=[['f1','a','b','c',1],
     ['f1','b','d','m',0],
     ['f2','a','b','c',0],
     ['f2','b','d','m',0],
     ['f3','a','b','c',1],
     ['f3','b','d','m',1],
     ['f4','a','b','c',0],
      ['f4','b','d','m',0]]

df= spark.createDataFrame(list,['function_name','param1','param2','param3','result'])
df.show()

+-------------+------+------+------+------+
|function_name|param1|param2|param3|result|
+-------------+------+------+------+------+
|           f1|     a|     b|     c|     1|
|           f1|     b|     d|     m|     0|
|           f2|     a|     b|     c|     0|
|           f2|     b|     d|     m|     0|
|           f3|     a|     b|     c|     1|
|           f3|     b|     d|     m|     1|
|           f4|     a|     b|     c|     0|
|           f4|     b|     d|     m|     0|
+-------------+------+------+------+------+

Группировка по имени функции, затем группировка по списку результатов (с использованием коллекционного списка), используя порядок param1, param2, param3 :

w=Window().partitionBy("function_name").orderBy(F.col("param1"),F.col("param2"),F.col("param3"))
w1=Window().partitionBy("function_name")
df1=df.withColumn("result_list", F.collect_list("result").over(w)).withColumn("result2",F.row_number().over(w))\
.withColumn("result3",F.max("result2").over(w1))\
.filter(F.col("result2")==F.col("result3")).drop("param1","param2","param3","result","result2","result3")


df1.groupBy("result_list")\
.agg(F.collect_list("function_name").alias("function_name_list")).show()

    +-----------+------------------+
    |result_list|function_name_list|
    +-----------+------------------+
    |     [1, 0]|              [f1]|
    |     [1, 1]|              [f3]|
    |     [0, 0]|          [f2, f4]|
    +-----------+------------------+

Для дальнейшего анализа, преобразования или очистки столбцов с типом массива я рекомендую вам проверить новые функции более высокого порядка в spark2.4 и выше.

(collect_list будет работать для spark1.6 и выше)

Функции высшего порядка в открытом коде:

https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark. sql .functions.collect_list

https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark. sql .functions.array_contains вперед

Релизы блоков данных: Ссылка: https://docs.databricks.com/delta/data-transformation/higher-order-lambda-functions.html