Столбец Pyspark делится на промежуточные итоги, сгруппированные по другому столбцу.

Question

Моя проблема похожа на это и это .В обоих сообщениях показано, как разделить значение столбца на общую сумму этого же столбца.В моем случае я хочу разделить значения столбца на сумму промежуточных итогов.Промежуточный итог рассчитывается путем группировки значений столбца в зависимости от другого столбца.Я немного изменяю пример, упомянутый в приведенных выше ссылках.

Вот мой фрейм данных

df = [[1,'CAT1',10], [2, 'CAT1', 11], [3, 'CAT2', 20], [4, 'CAT2', 22], [5, 'CAT3', 30]]
df = spark.createDataFrame(df, ['id', 'category', 'consumption'])
df.show()
+---+--------+-----------+
| id|category|consumption|
+---+--------+-----------+
|  1|    CAT1|         10|
|  2|    CAT1|         11|
|  3|    CAT2|         20|
|  4|    CAT2|         22|
|  5|    CAT3|         30|
+---+--------+-----------+

Я хочу разделить значение "потребления" на общее количество сгруппированных "категорий" и поместитьзначение в столбце «нормализовано», как показано ниже.

Промежуточные итоги необязательно должны быть в выводе (число 21, 42 и 30 в столбце потребления)

Чего я достиг к настоящему моменту df.crossJoin (

df.groupby('category').agg(F.sum('consumption').alias('sum_'))
).withColumn("normalized", F.col("consumption")/F.col("sum_"))\
.show()

+---+--------+-----------+--------+----+-------------------+
| id|category|consumption|category|sum_|         normalized|
+---+--------+-----------+--------+----+-------------------+
|  1|    CAT1|         10|    CAT2|  42|0.23809523809523808|
|  2|    CAT1|         11|    CAT2|  42| 0.2619047619047619|
|  1|    CAT1|         10|    CAT1|  21|0.47619047619047616|
|  2|    CAT1|         11|    CAT1|  21| 0.5238095238095238|
|  1|    CAT1|         10|    CAT3|  30| 0.3333333333333333|
|  2|    CAT1|         11|    CAT3|  30|0.36666666666666664|
|  3|    CAT2|         20|    CAT2|  42|0.47619047619047616|
|  4|    CAT2|         22|    CAT2|  42| 0.5238095238095238|
|  5|    CAT3|         30|    CAT2|  42| 0.7142857142857143|
|  3|    CAT2|         20|    CAT1|  21| 0.9523809523809523|
|  4|    CAT2|         22|    CAT1|  21| 1.0476190476190477|
|  5|    CAT3|         30|    CAT1|  21| 1.4285714285714286|
|  3|    CAT2|         20|    CAT3|  30| 0.6666666666666666|
|  4|    CAT2|         22|    CAT3|  30| 0.7333333333333333|
|  5|    CAT3|         30|    CAT3|  30|                1.0|
+---+--------+-----------+--------+----+-------------------+

Answer 1

Это еще один способ решения проблемы, предложенный OP, но без использования joins().

joins() в целом являются дорогостоящими операциями, и их следует избегать, когда это возможно.

# We first register our DataFrame as temporary SQL view
df.registerTempTable('table_view')
df = sqlContext.sql("""select id, category, 
                       consumption/sum(consumption) over (partition by category) as normalize
                       from table_view""")
df.show()
+---+--------+-------------------+
| id|category|          normalize|
+---+--------+-------------------+
|  3|    CAT2|0.47619047619047616|
|  4|    CAT2| 0.5238095238095238|
|  1|    CAT1|0.47619047619047616|
|  2|    CAT1| 0.5238095238095238|
|  5|    CAT3|                1.0|
+---+--------+-------------------+

Примечание: "" " используется для создания многострочных операторов для наглядности и аккуратности. С простым 'select id ....' это не сработает, если вы попытаетесь распространить Ваше утверждение в несколько строк. Нет необходимости говорить, что окончательный результат будет таким же.

Answer 2

Вы можете сделать то же самое, что и в ссылках, которые вы уже упоминали.Единственное отличие состоит в том, что вы должны рассчитать промежуточные итоги ранее с groupby и sum :

import pyspark.sql.functions as F
df = df.join(df.groupby('category').sum('consumption'), 'category')
df = df.select('id', 'category', F.round(F.col('consumption')/F.col('sum(consumption)'), 2).alias('normalized'))
df.show()

Выход:

+---+--------+----------+ 
| id|category|normalized| 
+---+--------+----------+ 
|  3|    CAT2|      0.48| 
|  4|    CAT2|      0.52| 
|  1|    CAT1|      0.48| 
|  2|    CAT1|      0.52| 
|  5|    CAT3|       1.0| 
+---+--------+----------+