Разделить столбец строки на основе разделителя и создать столбцы для каждого значения в Pyspark - PullRequest
Новая
       4

Разделить столбец строки на основе разделителя и создать столбцы для каждого значения в Pyspark

2 голосов
/ 25 января 2020

У меня есть тысячи файлов с данными в формате ниже 

a|b|c|clm4=1|clm5=3
a|b|c|clm4=9|clm6=60|clm7=23

, и я хочу прочитать его и преобразовать в кадр данных, как показано ниже 

clm1|clm2|clm3|clm4|clm5|clm6|clm7
a|b|c|1|3|null|null
a|b|c|9|null|60|23

Я попробовал метод ниже 

files = [f for f in glob.glob(pathToFile + "/**/*.txt.gz", recursive=True)]
df = spark.read.load(files, format='csv', sep = '|', header=None)

Но это дает мне результат ниже 

clm1, clm2, clm3, clm4, clm5
a, b, c, 1, 3
a, b, c, 9, null

1 Ответ

1 голос
/ 25 января 2020

Для Spark 2.4+ вы можете прочитать файлы как один столбец, а затем разделить их на |. Вы получите столбец массива, который вы можете преобразовать, используя функции высшего порядка : 

df.show(truncate=False)

+----------------------------+
|clm                         |
+----------------------------+
|a|b|c|clm4=1|clm5=3         |
|a|b|c|clm4=9|clm6=60|clm7=23|
+----------------------------+

Мы используем функцию transform для преобразования массива строк, который мы получаем при разбиении clm столбец в массив структур. Каждая структура содержит имя столбца, если оно присутствует (проверьте, содержит ли строка =) или назовите его clm + (i+1), где i - это его позиция. 

transform_expr = """
transform(split(clm, '[|]'), (x, i) -> 
                   struct(
                         IF(x like '%=%', substring_index(x, '=', 1), concat('clm', i+1)), 
                         substring_index(x, '=', -1)
                         )
        )
"""

Теперь используйте map_from_entries для преобразования массива в карту. И наконец, взорвите карту и развернитесь, чтобы получить столбцы 

df.select("clm", 
          explode(map_from_entries(expr(transform_expr))).alias("col_name", "col_value")
         ) \
  .groupby("clm").pivot('col_name').agg(first('col_value')) \
  .drop("clm") \
  .show(truncate=False)

Дает: 

+----+----+----+----+----+----+----+
|clm1|clm2|clm3|clm4|clm5|clm6|clm7|
+----+----+----+----+----+----+----+
|a   |b   |c   |9   |null|60  |23  |
|a   |b   |c   |1   |3   |null|null|
+----+----+----+----+----+----+----+
...