PySpark преобразовывает поле структуры внутри массива в строку

Question

У меня есть фрейм данных со схемой, подобной этой:

|-- order: string (nullable = true)
|-- travel: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- place: struct (nullable = true)
 |    |    |    |-- name: string (nullable = true)
 |    |    |    |-- address: string (nullable = true)
 |    |    |    |-- latitude: double (nullable = true)
 |    |    |    |-- longitude: double (nullable = true)
 |    |    |-- distance_in_kms: float (nullable = true)
 |    |    |-- estimated_time: struct (nullable = true)
 |    |    |    |-- seconds: long (nullable = true)
 |    |    |    |-- nanos: integer (nullable = true)

Я хочу получить секунды в estimated_time, преобразовать их в строку и объединить с s, а затем заменить * 1006.* с новым строковым значением.Например, { "seconds": "988", "nanos": "102" } будет преобразовано в 988s, поэтому схема изменится на

|-- order: string (nullable = true)
|-- travel: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- place: struct (nullable = true)
 |    |    |    |-- name: string (nullable = true)
 |    |    |    |-- address: string (nullable = true)
 |    |    |    |-- latitude: double (nullable = true)
 |    |    |    |-- longitude: double (nullable = true)
 |    |    |-- distance_in_kms: float (nullable = true)
 |    |    |-- estimated_time: string (nullable = true)

Как я могу это сделать в PySpark?

Более конкретный пример, я хочупреобразовать этот DF (визуализированный в JSON)

{
    "order": "c-331",
    "travel": [
        {
            "place": {
                "name": "A place",
                "address": "The address",
                "latitude": 0.0,
                "longitude": 0.0
            },
            "distance_in_kms": 1.0,
            "estimated_time": {
                "seconds": 988,
                "nanos": 102
            }
        }
    ]
}

в

{
    "order": "c-331",
    "travel": [
        {
            "place": {
                "name": "A place",
                "address": "The address",
                "latitude": 0.0,
                "longitude": 0.0
            },
            "distance_in_kms": 1.0,
            "estimated_time": "988s"
        }
    ]
}

Answer 1

Это можно сделать с помощью следующих функций pyspark:

• withColumn позволяет создать новый столбец.Мы будем использовать это для извлечения «оценочного времени»
• concat объединяет строковые столбцы
• lit создает столбец данной строки

Пожалуйста, взгляните на следующий пример:

from pyspark.sql import functions as F
j = '{"order":"c-331","travel":[{"place":{"name":"A place","address":"The address","latitude":0.0,"longitude":0.0},"distance_in_kms":1.0,"estimated_time":{"seconds":988,"nanos":102}}]}'
df = spark.read.json(sc.parallelize([j]))

#the following command creates a new column called estimated_time2 which contains the values of travel.estimated_time.seconds concatenated with a 's' 
bla = df.withColumn('estimated_time2', F.concat(df.travel.estimated_time.seconds[0].cast("string"), F.lit("s")))

#unfortunately it is currently not possible to use withColumn to add a new member to a struct. Therefore the following command replaces 'travel.estimated_time' with the before created column estimated_time2
bla = bla.select("order"
                , F.array(
                    F.struct(
                        bla.travel.distance_in_kms[0].alias("distance_in_kms")
                        ,bla.travel.place[0].alias("place")
                        , bla.estimated_time2.alias('estimated_time')
                        )).alias("travel"))

bla.show(truncate=False)
bla.printSchema()

И это вывод:

+-----+------------------------------------------+ 
|order|travel                                    | 
+-----+------------------------------------------+ 
|c-331|[[1.0,[The address,0.0,0.0,A place],988s]]| 
+-----+------------------------------------------+


root 
|-- order: string (nullable = true) 
|-- travel: array (nullable = false) 
| |-- element: struct (containsNull = false) 
| | |-- distance_in_kms: double (nullable = true)
| | |-- place: struct (nullable = true) 
| | | |-- address: string (nullable = true) 
| | | |-- latitude: double (nullable = true) 
| | | |-- longitude: double (nullable = true) 
| | | |-- name: string (nullable = true) 
| | |-- estimated_time: string (nullable = true)