Pyspark pivot для нескольких имен столбцов

Question

В настоящее время у меня есть фрейм данных df

id | c1   | c2   | c3 |
1  | diff | same | diff
2  | same | same | same
3  | diff | same | same
4  | same | same | same

Я хочу, чтобы мой вывод выглядел как

name| diff | same
c1  |   2  | 2
c2  |   0  | 4
c3  |   1  | 3

Когда я пытаюсь:

df.groupby('c2').pivot('c2').count() -> transformation A

|f2   | diff | same |
|same | null |  2
|diff | 2    |  null

Я предполагая, что мне нужно написать al oop для каждого столбца и передать его через преобразование A? Но у меня проблемы с получением права на преобразование. Пожалуйста, помогите

Answer 1

Pivot - дорогостоящая операция shuffle, и ее следует избегать , если возможно . Попробуйте использовать этот logi c с arrays_zip and explode до динамически сворачивать столбцы и groupby-aggregate.

from pyspark.sql import functions as F   

df.withColumn("cols", F.explode(F.arrays_zip(F.array([F.array(F.col(x),F.lit(x))\
                                                    for x in df.columns if x!='id']))))\
  .withColumn("name", F.col("cols.0")[1]).withColumn("val", F.col("cols.0")[0]).drop("cols")\
  .groupBy("name").agg(F.count(F.when(F.col("val")=='diff',1)).alias("diff"),\
                       F.count(F.when(F.col("val")=='same',1)).alias("same")).orderBy("name").show()

#+----+----+----+
#|name|diff|same|
#+----+----+----+
#|  c1|   2|   2|
#|  c2|   0|   4|
#|  c3|   1|   3|
#+----+----+----+

Вы также можете сделайте это с помощью exploding a map_type, создав map dynamically.

from pyspark.sql import functions as F
from itertools import chain

df.withColumn("cols", F.create_map(*(chain(*[(F.lit(name), F.col(name))\
                                  for name in df.columns if name!='id']))))\
  .select(F.explode("cols").alias("name","val"))\
  .groupBy("name").agg(F.count(F.when(F.col("val")=='diff',1)).alias("diff"),\
                       F.count(F.when(F.col("val")=='same',1)).alias("same")).orderBy("name").show()

#+----+----+----+
#|name|diff|same|
#+----+----+----+
#|  c1|   2|   2|
#|  c2|   0|   4|
#|  c3|   1|   3|
#+----+----+----+

Answer 2

from pyspark.sql.functions import *
df = spark.createDataFrame([(1,'diff','same','diff'),(2,'same','same','same'),(3,'diff','same','same'),(4,'same','same','same')],['idcol','C1','C2','C3'])
df.createOrReplaceTempView("MyTable")
#spark.sql("select * from MyTable").collect()
x1=spark.sql("select idcol, 'C1' AS col, C1 from MyTable union all select idcol, 'C2' AS col, C2 from MyTable  union all select idcol, 'C3' AS col, C3 from MyTable")
#display(x1)
x2=x1.groupBy('col').pivot('C1').agg(count('C1')).orderBy('col')
display(x2)