PySpark Dataframe приводит два столбца к новому столбцу значений на основе кортежей третьего столбца. - PullRequest
1 голос
/ 14 марта 2019

Как описывает предмет, у меня есть PySpark Dataframe, который мне нужен для преобразования двух столбцов в новый столбец, представляющий собой список кортежей на основе значения третьего столбца.Это приведение приведет к уменьшению или выравниванию кадра данных на значение ключа, идентификатор продукта в данном случае и результат по одной строке на ключ.

В этом кадре данных находятся сотни миллионов строк с 37M уникальных идентификаторов продукта,Поэтому мне нужен способ выполнить преобразование в искровом кластере без возврата каких-либо данных в драйвер (в данном случае Jupyter).

Вот выдержка из моего фрейма данных только для 1 продукта:

+-----------+-------------------+-------------+--------+----------+---------------+
| product_id|      purchase_date|days_warranty|store_id|year_month|       category|
+-----------+-------------------+-----------+----------+----------+---------------+
|02147465400|2017-05-16 00:00:00|           30|     205|   2017-05|     CATEGORY A|
|02147465400|2017-04-15 00:00:00|           30|     205|   2017-04|     CATEGORY A|
|02147465400|2018-07-11 00:00:00|           30|     205|   2018-07|     CATEGORY A|
|02147465400|2017-06-14 00:00:00|           30|     205|   2017-06|     CATEGORY A|
|02147465400|2017-03-16 00:00:00|           30|     205|   2017-03|     CATEGORY A|
|02147465400|2017-08-14 00:00:00|           30|     205|   2017-08|     CATEGORY A|
|02147465400|2017-09-12 00:00:00|           30|     205|   2017-09|     CATEGORY A|
|02147465400|2017-01-21 00:00:00|           30|     205|   2017-01|     CATEGORY A|
|02147465400|2018-08-14 00:00:00|           30|     205|   2018-08|     CATEGORY A|
|02147465400|2018-08-23 00:00:00|           30|     205|   2018-08|     CATEGORY A|
|02147465400|2017-10-11 00:00:00|           30|     205|   2017-10|     CATEGORY A|
|02147465400|2017-12-12 00:00:00|           30|     205|   2017-12|     CATEGORY A|
|02147465400|2017-02-15 00:00:00|           30|     205|   2017-02|     CATEGORY A|
|02147465400|2018-04-12 00:00:00|           30|     205|   2018-04|     CATEGORY A|
|02147465400|2018-03-12 00:00:00|           30|     205|   2018-03|     CATEGORY A|
|02147465400|2018-05-15 00:00:00|           30|     205|   2018-05|     CATEGORY A|
|02147465400|2018-02-12 00:00:00|           30|     205|   2018-02|     CATEGORY A|
|02147465400|2018-06-14 00:00:00|           30|     205|   2018-06|     CATEGORY A|
|02147465400|2018-01-11 00:00:00|           30|     205|   2018-01|     CATEGORY A|
|02147465400|2017-07-20 00:00:00|           30|     205|   2017-07|     CATEGORY A|
|02147465400|2017-11-11 00:00:00|           30|     205|   2017-11|     CATEGORY A|
|02147465400|2017-01-05 00:00:00|           90|     205|   2017-01|     CATEGORY B|
|02147465400|2017-01-21 00:00:00|           90|     205|   2017-01|     CATEGORY B|
|02147465400|2017-10-09 00:00:00|           90|     205|   2017-10|     CATEGORY B|
|02147465400|2018-07-11 00:00:00|           90|     205|   2018-07|     CATEGORY B|
|02147465400|2017-04-16 00:00:00|           90|     205|   2017-04|     CATEGORY B|
|02147465400|2018-09-16 00:00:00|           90|     205|   2018-09|     CATEGORY B|
|02147465400|2018-04-14 00:00:00|           90|     205|   2018-04|     CATEGORY B|
|02147465400|2018-01-12 00:00:00|           90|     205|   2018-01|     CATEGORY B|
|02147465400|2017-07-15 00:00:00|           90|     205|   2017-07|     CATEGORY B|
+-----------+-------------------+-----------+----------+----------+---------------+

Вот требуемый результирующий фрейм данных, одна строка для одного продукта, где строки исходного фрейма данных имеют столбцы purchase_date и days_warranty, приведенные в виде массива кортежей в новые столбцы на основе значения столбца категории:

+-----------+----------------------------+----------------------------+
| product_id|                  CATEGORY A|                  CATEGORY B| 
+-----------+----------------------------+----------------------------+
|02147465400| [ (2017-05-16 00:00:00,30),| [ (2017-01-05 00:00:00,90),| 
|           |   (2017-04-15 00:00:00,30),|   (2017-01-21 00:00:00,90),|
|           |   (2018-07-11 00:00:00,30),|   (2017-10-09 00:00:00,90),|
|           |   (2017-06-14 00:00:00,30),|   (2018-07-11 00:00:00,90),|
|           |   (2017-03-16 00:00:00,30),|   (2017-04-16 00:00:00,90),|
|           |   (2017-08-14 00:00:00,30),|   (2018-09-16 00:00:00,90),|
|           |   (2017-09-12 00:00:00,30),|   (2018-04-14 00:00:00,90),|
|           |   (2017-01-21 00:00:00,30),|   (2018-01-12 00:00:00,90),|
|           |   (2018-08-14 00:00:00,30),|   (2017-07-15 00:00:00,90) |
|           |   (2018-08-23 00:00:00,30),| ]                          |
|           |   (2017-10-11 00:00:00,30),|                            |
|           |   (2017-12-12 00:00:00,30),|                            |
|           |   (2017-02-15 00:00:00,30),|                            |
|           |   (2018-04-12 00:00:00,30),|                            |
|           |   (2018-03-12 00:00:00,30),|                            |
|           |   (2018-05-15 00:00:00,30),|                            |
|           |   (2018-02-12 00:00:00,30),|                            |
|           |   (2018-06-14 00:00:00,30),|                            |
|           |   (2018-01-11 00:00:00,30),|                            |
|           |   (2017-07-20 00:00:00,30) |                            |
|           | ]                                                       |
+-----------+----------------------------+----------------------------+

Ответы [ 2 ]

1 голос
/ 20 марта 2019

В случае, если у вас есть проблемы с производительностью в pivot, приведенный ниже подход является еще одним решением той же проблемы, хотя он позволяет вам иметь больший контроль, разбивая работу на фазы для каждой категории с помощью цикла for. Для каждой итерации это добавит новые данные для category_x в acc_df, в котором будут храниться накопленные результаты.

schema = ArrayType( 
        StructType((  
            StructField("p_date", StringType(), False), 
            StructField("d_warranty", StringType(), False)  
        )) 
    )

    tuple_list_udf = udf(tuple_list, schema)

    buf_size = 5 # if you get OOM error decrease this to persist more often

    categories = df.select("category").distinct().collect()

    acc_df = spark.createDataFrame(sc.emptyRDD(), df.schema) # create an empty df which holds the accumulated results for each category

    for idx, c in enumerate(categories):
        col_name = c[0].replace(" ", "_") # spark complains for columns containing space
        cat_df = df.where(df["category"] == c[0]) \
                .groupBy("product_id") \
                .agg(
                    F.collect_list(F.col("purchase_date")).alias("p_date"), 
                    F.collect_list(F.col("days_warranty")).alias("d_warranty")) \
                .withColumn(col_name, tuple_list_udf(F.col("p_date"), F.col("d_warranty"))) \
                .drop("p_date", "d_warranty")

        if idx == 0:
            acc_df = cat_df
        else:
            acc_df = acc_df \
                .join(cat_df.alias("cat_df"), "product_id") \
                .drop(F.col("cat_df.product_id"))

        # you can persist here every buf_size iterations
        if idx + 1 % buf_size == 0:
            acc_df = acc_df.persist()

Функция tuple_list отвечает за создание списка с кортежами из столбцов purchase_date и days_warranty.

def tuple_list(pdl, dwl):
    return list(zip(pdl, dwl))

Вывод этого будет:

+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|product_id |CATEGORY_B                                                                                                                                                                                                                                         |CATEGORY_A                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|02147465400|[[2017-04-16 00:00:00, 90], [2018-09-16 00:00:00, 90], [2017-10-09 00:00:00, 90], [2018-01-12 00:00:00, 90], [2018-07-11 00:00:00, 90], [2017-01-21 00:00:00, 90], [2018-04-14 00:00:00, 90], [2017-01-05 00:00:00, 90], [2017-07-15 00:00:00, 90]]|[[2017-06-14 00:00:00, 30], [2018-08-14 00:00:00, 30], [2018-01-11 00:00:00, 30], [2018-04-12 00:00:00, 30], [2017-10-11 00:00:00, 30], [2017-05-16 00:00:00, 30], [2018-05-15 00:00:00, 30], [2017-04-15 00:00:00, 30], [2017-02-15 00:00:00, 30], [2018-02-12 00:00:00, 30], [2017-01-21 00:00:00, 30], [2018-07-11 00:00:00, 30], [2018-06-14 00:00:00, 30], [2017-03-16 00:00:00, 30], [2017-07-20 00:00:00, 30], [2018-08-23 00:00:00, 30], [2017-09-12 00:00:00, 30], [2018-03-12 00:00:00, 30], [2017-12-12 00:00:00, 30], [2017-08-14 00:00:00, 30], [2017-11-11 00:00:00, 30]]|
+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
0 голосов
/ 14 марта 2019

Предполагается, что ваш Dataframe называется df:

from pyspark.sql.functions import struct
from pyspark.sql.functions import collect_list

gdf = (df.select("product_id", "category", struct("purchase_date", "warranty_days").alias("pd_wd"))
.groupBy("product_id")
.pivot("category")
.agg(collect_list("pd_wd")))

По сути, вы должны сгруппировать purchase_date и warranty_days в один столбец, используя struct().Затем вы просто группируете по product_id, поворачиваетесь по category, можете агрегировать как collect_list().

...