столбец изменения pyspark dataframe с двумя массивами в столбцы

Question

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

Хорошо, подробности.

Среда - pyspark 2.3.2 и python 2.7

Пример столбца содержит 2 массива, которые соотносятся друг с другом от 1 до 1. Я хотел бы создать столбец для каждого значения в массиве заголовки и поставить соответствующее имя (в персона массив) соответствующего столбца.

Я придумал пример, чтобы сосредоточиться на моей проблеме с изменением кадра данных.

import json
from pyspark.sql.types import ArrayType, StructType, StructField, StringType
from pyspark.sql import functions as f

input = { "sample": {    "titles": ["Engineer", "Designer", "Manager"],    "person": ["Mary", "Charlie", "Mac"]  },  "location": "loc a"},{ "sample": {    "titles": ["Engineer", "Owner"],
    "person": ["Tom", "Sue"]  },  "location": "loc b"},{ "sample": {    "titles": ["Engineer", "Designer"],    "person": ["Jane", "Bill"]  },  "location": "loc a"}

a = [json.dumps(input)]
jsonRDD = sc.parallelize(a)
df = spark.read.json(jsonRDD)

Это схема моего фрейма данных:

In [4]: df.printSchema()
root
 |-- location: string (nullable = true)
 |-- sample: struct (nullable = true)
 |    |-- person: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- titles: array (nullable = true)
 |    |    |-- element: string (containsNull = true)

Данные моего фрейма данных:

In [5]: df.show(truncate=False)
+--------+-----------------------------------------------------+
|location|sample                                               |
+--------+-----------------------------------------------------+
|loc a   |[[Mary, Charlie, Mac], [Engineer, Designer, Manager]]|
|loc b   |[[Sue, Tom], [Owner, Engineer]]                      |
|loc a   |[[Jane, Bill], [Engineer, Designer]]                 |
+--------+-----------------------------------------------------+

И как бы я хотел, чтобы мой фрейм данных выглядел:

+--------+-----------------------------------------------------+------------+-----------+---------+---------+
|location|sample                                               |Engineer    |Desginer   |Manager  | Owner   |
+--------+-----------------------------------------------------+------------+-----------+---------+---------+
|loc a   |[[Mary, Charlie, Mac], [Engineer, Designer, Manager]]|Mary        |Charlie    |Mac      |         |
|loc b   |[[Sue, Tom], [Owner, Engineer]]                      |Tom         |           |         |Sue      |
|loc a   |[[Jane, Bill], [Engineer, Designer]]                 |Jane        |Bill       |         |         |
+--------+-----------------------------------------------------+------------+-----------+---------+---------+

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

Answer 1

@ user10601094 помог мне получить ответ на этот вопрос. Ниже выложено полное решение, чтобы помочь кому-то, у кого может быть похожий вопрос

Я не очень хорошо говорю на python, поэтому, пожалуйста, не стесняйтесь предлагать лучшие подходы

In [1]: import json
   ...: from pyspark.sql import functions as f
   ...: 

In [2]: # define a sample data set
   ...: input = { "sample": {    "titles": ["Engineer", "Designer", "Manager"],    "person": ["Mary", "Charlie", "Mac"]  },  "location": "loc a"},{ "sample": {    "titles": ["Engineer", "Owner"],
   ...:     "person": ["Tom", "Sue"]  },  "location": "loc b"},{ "sample": {    "titles": ["Engineer", "Designer"],    "person": ["Jane", "Bill"]  },  "location": "loc a"}

In [3]: # create a dataframe with the sample json data
   ...: a = [json.dumps(input)]
   ...: jsonRDD = sc.parallelize(a)
   ...: df = spark.read.json(jsonRDD)
   ...: 
2018-11-03 20:48:09 WARN  ObjectStore:568 - Failed to get database global_temp, returning NoSuchObjectException

In [4]: # Change the array in the sample column to a dictionary
   ...: # swap the columns so the titles are the key
   ...: 
   ...: # UDF to convert 2 arrays into a map
   ...: @f.udf("map<string,string>")
   ...: def as_dict(x):
   ...:     return dict(zip(x[1],x[0])) if x else None
   ...: 

In [5]: # create a new dataframe based on the original dataframe
   ...: dfmap = df.withColumn("sample", as_dict("sample"))

In [6]: # Convert sample column to be title columns based on the map
   ...: 
   ...: # get the columns names, stored in the keys
   ...: keys = (dfmap
   ...:     .select(f.explode("sample"))
   ...:     .select("key")
   ...:     .distinct()
   ...:     .rdd.flatMap(lambda x: x)
   ...:     .collect())

In [7]: # create a list of column names 
   ...: exprs = [f.col("sample").getItem(k).alias(k) for k in keys]
   ...: 

In [8]: dfmap.select(dfmap.location, *exprs).show()
+--------+--------+--------+-------+-----+
|location|Designer|Engineer|Manager|Owner|
+--------+--------+--------+-------+-----+
|   loc a| Charlie|    Mary|    Mac| null|
|   loc b|    null|     Tom|   null|  Sue|
|   loc a|    Bill|    Jane|   null| null|
+--------+--------+--------+-------+-----+

Answer 2

1. Без explode

   • Сначала преобразуйте каждый struct в map:

     from pyspark.sql.functions import udf
     
     @udf("map<string,string>")
     def as_dict(x):
         return dict(zip(*x)) if x else None
     
     dfmap = df.withColumn("sample", as_dict("sample")
     

   • Затем используйте метод, показанный в PySpark, преобразующий столбец типа 'map' в несколько столбцов в кадре данных , чтобы разбить карту на столбцы

2. С explode

   • Добавить уникальный идентификатор, используя monotonically_increasing_id.
   • Используйте один из методов, показанных в Pyspark: разбить несколько столбцов массива на строки , чтобы разбить оба массива вместе, или explode map, созданный с помощью метода first.
   • pivot результат, группировка по добавленному идентификатору и другим полям, которые вы хотите сохранить, поворот по title и взятие first(person)