Question

У меня есть образец входного кадра данных, как показано ниже, но столбцы значений (clm, начинающиеся с m) могут быть n числом. Также я использовал customer_id в качестве первичного ключа (но у меня больше не может быть # первичного ключа на основе входных данных).

customer_id|month_id|m1    |m2 |m3 ....to....m_n
1001      |  01    |10     |20    
1002      |  01    |20     |30    
1003      |  01    |30     |40
1001      |  02    |40     |50    
1002      |  02    |50     |60    
1003      |  02    |60     |70
1001      |  03    |70     |80    
1002      |  03    |80     |90    
1003      |  03    |90     |100

Теперь, основываясь на столбцах входных значений, я должен вычислить новые столбцы на основе совокупной суммы или среднего значения. Рассмотрим пример:

cumulative sum on [m1, ......, m10] and 
cumulative avg on [m11, ......., m20] columns

Исходя из этого, мне нужно вычислить новые столбцы. Я пробовал его на основе функции windows и смог вычислить новые столбцы. Но моя проблема в том, что из-за размера данных я выполняю вычисления один за другим с обновленным фреймом данных с новыми столбцами.

Моя попытка:

a = [m1, ......, m10]
b = [m11, ......, m20]
rnum = (Window.partitionBy("partner_id").orderBy("month_id").rangeBetween(Window.unboundedPreceding, 0))
for item in a:
   var = n
   df = df.withColumn(var + item[1:], F.sum(item).over(rnum))
for item in b:
   var = n
   df = df.withColumn(var + item[1:], F.avg(item).over(rnum))

Вывод data:

customer_id|month_id|m1     |m2    |m11     |m12   |n1   |n2  |n11  |n12
1001       |  01    |10     |20    |10      |20    |10   |20  |10   |20
1002       |  01    |20     |30    |10      |20    |20   |30  |10   |20
1003       |  01    |30     |40    |10      |20    |30   |40  |10   |20
1001       |  02    |40     |50    |10      |20    |50   |35  |10   |20
1002       |  02    |50     |60    |10      |20    |70   |55  |10   |20
1003       |  02    |60     |70    |10      |20    |90   |75  |10   |20
1001       |  03    |70     |80    |10      |20    |120  |75  |10   |20
1002       |  03    |80     |90    |10      |20    |150  |105 |10   |20
1003       |  03    |90     |100   |10      |20    |180  |135 |10   |20

Но можем ли мы сделать ту же операцию, разделив фрейм данных на два, со столбцами кумулятивной суммы в одном и кумулятивными столбцами avg в другом фрейме данных вместе с первичным ключом, а затем выполните операция затем клуб рассчитанный фрейм данных ????

sathiyarajan · Answer 1 · 04 августа 2020

Логический план, оптимизированный для подхода DF1

== Optimized Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Project [m1#15, m2#16, sum1#27, sum2#38, customer_id#5334, month_id#5335, m3#5338, m4#5339, avg3#465, avg4#474]
      +- Join Inner, ((customer_id#13 = customer_id#5334) && (month_id#14 = month_id#5335))
         :- Project [customer_id#13, month_id#14, m1#15, m2#16, sum1#27, sum2#38]
         :  +- Filter isnotnull(month_id#14)
         :     +- Window [sum(_w0#39) windowspecdefinition(customer_id#13, month_id#14 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum2#38], [customer_id#13], [month_id#14 ASC NULLS FIRST]
         :        +- Project [customer_id#13, month_id#14, m1#15, m2#16, sum1#27, cast(m2#16 as double) AS _w0#39]
         :           +- Window [sum(_w0#28) windowspecdefinition(customer_id#13, month_id#14 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum1#27], [customer_id#13], [month_id#14 ASC NULLS FIRST]
         :              +- Project [customer_id#13, month_id#14, m1#15, m2#16, cast(m1#15 as double) AS _w0#28]
         :                 +- Filter isnotnull(customer_id#13)
         :                    +- LogicalRDD [customer_id#13, month_id#14, m1#15, m2#16, m3#17, m4#18]
         +- Project [customer_id#5334, month_id#5335, m3#5338, m4#5339, avg3#465, avg4#474]
            +- Filter isnotnull(month_id#5335)
               +- Window [avg(_w0#475) windowspecdefinition(customer_id#5334, month_id#5335 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg4#474], [customer_id#5334], [month_id#5335 ASC NULLS FIRST]
                  +- Project [customer_id#5334, month_id#5335, m3#5338, m4#5339, avg3#465, cast(m4#5339 as double) AS _w0#475]
                     +- Window [avg(_w0#466) windowspecdefinition(customer_id#5334, month_id#5335 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg3#465], [customer_id#5334], [month_id#5335 ASC NULLS FIRST]
                        +- Project [customer_id#5334, month_id#5335, m3#5338, m4#5339, cast(m3#5338 as double) AS _w0#466]
                           +- Filter isnotnull(customer_id#5334)
                              +- LogicalRDD [customer_id#5334, month_id#5335, m1#5336, m2#5337, m3#5338, m4#5339]

Логический план, оптимизированный для подхода DF

== Optimized Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, sum1#14, sum2#25, avg3#447, avg4#460]
      +- Window [avg(_w0#461) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg4#460], [customer_id#0], [month_id#1 ASC NULLS FIRST]
         +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, sum1#14, sum2#25, avg3#447, cast(m4#5 as double) AS _w0#461]
            +- Window [avg(_w0#448) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS avg3#447], [customer_id#0], [month_id#1 ASC NULLS FIRST]
               +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, sum1#14, sum2#25, cast(m3#4 as double) AS _w0#448]
                  +- Window [sum(_w0#26) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum2#25], [customer_id#0], [month_id#1 ASC NULLS FIRST]
                     +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, sum1#14, cast(m2#3 as double) AS _w0#26]
                        +- Window [sum(_w0#15) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum1#14], [customer_id#0], [month_id#1 ASC NULLS FIRST]
                           +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, cast(m1#2 as double) AS _w0#15]
                              +- LogicalRDD [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5]

Если вы видите вышеуказанное DF Approach Optimized Logical Plan, У него есть план расчета SUM во время расчета AVG, который может быть неэффективным.

+- Window [sum(_w0#26) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum2#25], [customer_id#0], [month_id#1 ASC NULLS FIRST]
                     +- Project [customer_id#0, month_id#1, m1#2, m2#3, m3#4, m4#5, sum1#14, cast(m2#3 as double) AS _w0#26]
                        +- Window [sum(_w0#15) windowspecdefinition(customer_id#0, month_id#1 ASC NULLS FIRST, RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum1#14],

Вы можете сузить размер фрейма данных, когда это возможно, и продолжить вычисления. В то же время план join был добавлен для двух наборов данных в Оптимизированный логический план DF1. Во многих случаях соединения всегда выполняются медленно, поэтому лучше попытаться настроить среду выполнения Spark Engine с помощью:

code - repartition & cache
configs - executor, driver, memoryOverhead, number of cores

код, который я пробовал с m1,m2,m3,m4 столбцами.

# pyspark --driver-memory 1G --executor-memory 2G --executor-cores 1 --num-executors 1
from pyspark.sql import Row
import pyspark.sql.functions as F
from pyspark.sql.window import Window

drow = Row("customer_id","month_id","m1","m2","m3","m4")

data=[drow("1001","01","10","20","10","20"),drow("1002","01","20","30","20","30"),drow("1003","01","30","40","30","40"),drow("1001","02","40","50","40","50"),drow("1002","02","50","60","50","60"),drow("1003","02","60","70","60","70"),drow("1001","03","70","80","70","80"),drow("1002","03","80","90","80","90"),drow("1003","03","90","100","90","100")]

df = spark.createDataFrame(data)

df1=df.select("customer_id","month_id","m3","m4")

a = ["m1","m2"]
b = ["m3","m4"]
rnum = (Window.partitionBy("customer_id").orderBy("month_id").rangeBetween(Window.unboundedPreceding, 0))
for item in a:
    var = "sum"
    df = df.withColumn(var + item[1:], F.sum(item).over(rnum))
df.show()
'''
+-----------+--------+---+---+---+---+-----+-----+
|customer_id|month_id| m1| m2| m3| m4| sum1| sum2|
+-----------+--------+---+---+---+---+-----+-----+
|       1003|      01| 30| 40| 30| 40| 30.0| 40.0|
|       1003|      02| 60| 70| 60| 70| 90.0|110.0|
|       1003|      03| 90|100| 90|100|180.0|210.0|
|       1002|      01| 20| 30| 20| 30| 20.0| 30.0|
|       1002|      02| 50| 60| 50| 60| 70.0| 90.0|
|       1002|      03| 80| 90| 80| 90|150.0|180.0|
|       1001|      01| 10| 20| 10| 20| 10.0| 20.0|
|       1001|      02| 40| 50| 40| 50| 50.0| 70.0|
|       1001|      03| 70| 80| 70| 80|120.0|150.0|
+-----------+--------+---+---+---+---+-----+-----+
'''
for item in b:
    var = "avg"
    df = df.withColumn(var + item[1:], F.avg(item).over(rnum))
df.show()

'''
+-----------+--------+---+---+---+---+-----+-----+----+----+
|customer_id|month_id| m1| m2| m3| m4| sum1| sum2|avg3|avg4|
+-----------+--------+---+---+---+---+-----+-----+----+----+
|       1003|      01| 30| 40| 30| 40| 30.0| 40.0|30.0|40.0|
|       1003|      02| 60| 70| 60| 70| 90.0|110.0|45.0|55.0|
|       1003|      03| 90|100| 90|100|180.0|210.0|60.0|70.0|
|       1002|      01| 20| 30| 20| 30| 20.0| 30.0|20.0|30.0|
|       1002|      02| 50| 60| 50| 60| 70.0| 90.0|35.0|45.0|
|       1002|      03| 80| 90| 80| 90|150.0|180.0|50.0|60.0|
|       1001|      01| 10| 20| 10| 20| 10.0| 20.0|10.0|20.0|
|       1001|      02| 40| 50| 40| 50| 50.0| 70.0|25.0|35.0|
|       1001|      03| 70| 80| 70| 80|120.0|150.0|40.0|50.0|
+-----------+--------+---+---+---+---+-----+-----+----+----+
'''

for item in b:
    var = "avg"
    df1 = df1.withColumn(var + item[1:], F.avg(item).over(rnum))

'''
+-----------+--------+---+---+----+----+
|customer_id|month_id| m3| m4|avg3|avg4|
+-----------+--------+---+---+----+----+
|       1003|      01| 30| 40|30.0|40.0|
|       1003|      02| 60| 70|45.0|55.0|
|       1003|      03| 90|100|60.0|70.0|
|       1002|      01| 20| 30|20.0|30.0|
|       1002|      02| 50| 60|35.0|45.0|
|       1002|      03| 80| 90|50.0|60.0|
|       1001|      01| 10| 20|10.0|20.0|
|       1001|      02| 40| 50|25.0|35.0|
|       1001|      03| 70| 80|40.0|50.0|
+-----------+--------+---+---+----+----+
'''
#join the DFs after DF1 avg & DF sum calculation.

df2=df.join(df1,(df1.customer_id == df.customer_id)& (df1.month_id == df.month_id)).drop(df.m3).drop(df.m4).drop(df1.month_id).drop(df1.customer_id)

'''
df2.show()
+---+---+-----+-----+-----------+--------+---+---+----+----+
| m1| m2| sum1| sum2|customer_id|month_id| m3| m4|avg3|avg4|
+---+---+-----+-----+-----------+--------+---+---+----+----+
| 10| 20| 10.0| 20.0|       1001|      01| 10| 20|10.0|20.0|
| 70| 80|120.0|150.0|       1001|      03| 70| 80|40.0|50.0|
| 40| 50| 50.0| 70.0|       1001|      02| 40| 50|25.0|35.0|
| 80| 90|150.0|180.0|       1002|      03| 80| 90|50.0|60.0|
| 50| 60| 70.0| 90.0|       1002|      02| 50| 60|35.0|45.0|
| 20| 30| 20.0| 30.0|       1002|      01| 20| 30|20.0|30.0|
| 30| 40| 30.0| 40.0|       1003|      01| 30| 40|30.0|40.0|
| 90|100|180.0|210.0|       1003|      03| 90|100|60.0|70.0|
| 60| 70| 90.0|110.0|       1003|      02| 60| 70|45.0|55.0|
+---+---+-----+-----+-----------+--------+---+---+----+----+
'''

Mohan · Answer 2 · 04 августа 2020

Исходя из вашего вопроса, насколько я понимаю, вы пытаетесь разделить операцию для параллельного выполнения задач и для экономии времени.

Вам не нужно распараллеливать выполнение, поскольку выполнение будет автоматически распараллеливается в Spark при выполнении любых операций, таких как collect (), show (), count (), запись в созданный вами фрейм данных. Это происходит из-за ленивого выполнения Spark

Если вы по-прежнему хотите разделить операции по какой-то другой причине, вы можете использовать потоки. В статье ниже вы найдете дополнительную информацию о потоковой передаче в pyspark: https://medium.com/@everisUS / thread-in-pyspark-a6e8005f6017

Как разделить фрейм данных pyspark и создать новые столбцы

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

Ответы [ 2 ]

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Как разделить фрейм данных pyspark и создать новые столбцы

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

Ответы [ 2 ]

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Похожие темы