DataPrame pyspark транспонировать несколько столбцов на основе суффикса столбца - PullRequest
Новая
       17

DataPrame pyspark транспонировать несколько столбцов на основе суффикса столбца

0 голосов
/ 17 апреля 2019

имеют фрейм данных (столбцы с до e имеют до 15 вариантов) 

cola, colb, colc_1, cold_1, cole_1, colc_2, cold_2, cole_2...
1,     2,     3,     4,      5,      6,      7,      8

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

cola, colb, new_col colc, cold, cole, 
1,     2,     _1,     3,    4,     5
1,     2,     _2,     6,    7,     8

Ищем транспонировать colc для cole и использовать суффикс этих столбцов(_1, _2 ..._ 15) как значение транспонированного поля (new_col)

Я могу сделать это в Pandas, используя melt и pivot, но кадр данных в этом примере слишком велик для преобразования вPandas DF и должно быть сделано в Pyspark или AWS клей

1 Ответ

1 голос
/ 18 апреля 2019

Вы можете попробовать select() и union().код ниже сначала перечисляет базовую логику, а затем использует функцию reduce() для удаления всех промежуточных кадров данных: 

from pyspark.sql import functions as F
from functools import reduce

df = spark.createDataFrame([
        (1,2,3,4,5,6,7,8)
      , (11,12,13,14,15,16,17,18)
      , (21,22,23,24,25,26,27,28)
    ],
    [   'cola', 'colb'
      , 'colc_1', 'cold_1', 'cole_1'
      , 'colc_2', 'cold_2', 'cole_2'
    ])

# create df1 with all columns for new_col = '_1'
df1 = df.select('cola', 'colb', F.lit('_1'), 'colc_1', 'cold_1', 'cole_1')

df1.show()
#+----+----+---+------+------+------+
#|cola|colb| _1|colc_1|cold_1|cole_1|
#+----+----+---+------+------+------+
#|   1|   2| _1|     3|     4|     5|
#|  11|  12| _1|    13|    14|    15|
#|  21|  22| _1|    23|    24|    25|
#+----+----+---+------+------+------+

# do the similar for '_2'
df2 = df.select('cola', 'colb', F.lit('_2'), *["col{}_2".format(i) for i in list("cde")])
#+----+----+---+------+------+------+
#|cola|colb| _2|colc_2|cold_2|cole_2|
#+----+----+---+------+------+------+
#|   1|   2| _2|     6|     7|     8|
#|  11|  12| _2|    16|    17|    18|
#|  21|  22| _2|    26|    27|    28|
#+----+----+---+------+------+------+

# then union these two dataframe and adjust the final column names
df_new = df1.union(df2).toDF('cola', 'colb', 'new_col', 'colc', 'cold', 'cole')
df_new.show()
#+----+----+-------+----+----+----+
#|cola|colb|new_col|colc|cold|cole|
#+----+----+-------+----+----+----+
#|   1|   2|     _1|   3|   4|   5|
#|  11|  12|     _1|  13|  14|  15|
#|  21|  22|     _1|  23|  24|  25|
#|   1|   2|     _2|   6|   7|   8|
#|  11|  12|     _2|  16|  17|  18|
#|  21|  22|     _2|  26|  27|  28|
#+----+----+-------+----+----+----+

Затем мы можем использовать функцию reduce() для обработки всех групп столбцов без вышеуказанного временного df1, df2и т.д.: 

# setup the list of columns to be normalized
normalize_cols = ["col{}".format(c) for c in list("cde")]
# ["colc", "cold", "cole"]    

# change N to 16 to cover new_col from '_1' to '_15'
N = 3

# use reduce to handle all groups
df_new = reduce(
    lambda d1,d2: d1.union(d2)
  , [ df.select('cola', 'colb', F.lit('_{}'.format(i)), *["{}_{}".format(c,i) for c in normalize_cols]) for i in range(1,N) ]
).toDF('cola', 'colb', 'new_col', *normalize_cols)

Другой способ - использовать F.array() и F.explode() (используйте limit () для всех _N): 

df.withColumn('d1', F.array(F.lit('_1'), *['col{}_1'.format(c) for c in list("cde")])) \
  .withColumn('d2', F.array(F.lit('_2'), *['col{}_2'.format(c) for c in list("cde")])) \
  .withColumn('h', F.array('d1', 'd2')) \
  .withColumn('h1', F.explode('h')) \
  .select('cola', 'colb', *[ F.col('h1')[i] for i in range(4)]) \
  .toDF('cola', 'colb', 'new_col', 'colc', 'cold', 'cole') \
  .show()

Обновление за комментарий:

Для денормализации кадра данных я использую F.array(), а затем F.collect_list, чтобы сгруппировать столбцы в список массивов, а затем сослаться на значения из результата groupby():

Использование функции Window для установки порядка элементов в collect_list: ссылочная ссылка 

N = 3
normalize_cols = ["col{}".format(c) for c in list("cde")]

# win spec so that element in collect_list are sorted based on 'new_col'
win = Window.partitionBy('cola', 'colb').orderBy('new_col')

df_new.withColumn('cols', F.array(normalize_cols)) \
      .withColumn('clist', F.collect_list('cols').over(win)) \
      .groupby('cola', 'colb').agg(F.last('clist').alias('clist1')) \
      .select('cola', 'colb', *[ F.col('clist1')[i].alias('c{}'.format(i)) for i in range(N-1)]) \
      .select('cola', 'colb', *[ F.col('c{}'.format(i))[j].alias('{}_{}'.format(normalize_cols[j],i+1)) for i in range(N-1) for j in range(len(normalize_cols)) ]) \
      .show()    

# +----+----+------+------+------+------+------+------+                           
# |cola|colb|colc_1|cold_1|cole_1|colc_2|cold_2|cole_2|
# +----+----+------+------+------+------+------+------+
# |  11|  12|    13|    14|    15|    16|    17|    18|
# |  21|  22|    23|    24|    25|    26|    27|    28|
# |   1|   2|     3|     4|     5|     6|     7|     8|
# +----+----+------+------+------+------+------+------+

Некоторые пояснения:

  • F.last() в groupby.agg () возвращает полный коллекционный список из функции Window в том же разделеBy (groupby)
  • 1-й select() конвертирующий collect_list () в c0 , c1
  • 2-й select() преобразовать c0 в colc_1, cold_1, cole_1 и c1 до colc_2, cold_2, cole_2
...