PySpark: длина объекта не совпадает с длиной полей – создание новой схемы ⇐ Python

[Anonymous]

В настоящее время я работаю с AWS Glue и PySpark. Я пытаюсь вручную создать свою схему Spark и применить ее к фрейму данных, чтобы исправить некоторые проблемы, возникающие у меня с некоторыми столбцами.
Моя текущая проблема связана со столбцами со структурным типом. Мои данные содержат только 6 полей, представленных в JSON, считанных через Glue, но в будущем они могут иметь 8 полей. Если я создаю свою схему с 6 полями, которые я получил на данный момент, она работает нормально, но если я создаю схему с 8 полями, которые я должен получить, я получаю следующую ошибку: ValueError: имя поля_структура: длина объекта (6) не совпадает с длиной полей (8).
Как лучше всего применить обновленную схему? Я думаю, что мне следует применить функцию к столбцу структуры и создать недостающие ключи с нулевыми значениями, чтобы в моих данных было ожидаемых 8 ключей. Есть идеи, как это сделать? Если я это сделаю, я думаю, что для всех столбцов типа структуры мне следует проанализировать схему, сравнить ключи между данными в столбце (asDict()?) и схемой и, наконец, добавить недостающие. У меня такое ощущение, что есть лучший способ использовать Spark или, может быть, какие-то параметры, которые я еще не нашел, чтобы автоматически исправить это при создании кадра данных?

Код: Выделить всё

import findspark

findspark.init()

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType, IntegerType, StructType, StructField

spark = SparkSession.builder.master("local[1]") \
.appName('test_schema') \
.getOrCreate()

data = [
(32, ("James", "Smith"), ("street", 42)),
(18, ("Nina", "Smith"), ("street 2", 12))
]

schema = StructType([
StructField('age', IntegerType(), True),
StructField('name_struct', StructType([
StructField('first_name', StringType(), True),
StructField('last_name', StringType(), True)
])),
StructField('address_struct', StructType([
StructField('street', StringType(), True),
StructField('number', IntegerType(), True)
]))
])

df = spark.createDataFrame(data=data, schema=schema)
df.printSchema()
df.show()

schema2 = StructType([
StructField('age', IntegerType(), True),
StructField('name_struct', StructType([
StructField('first_name', StringType(), True),
StructField('last_name', StringType(), True),
StructField('middle_name', StringType(), True)
])),
StructField('address_struct', StructType([
StructField('street', StringType(), True),
StructField('number', IntegerType(), True),
StructField('zip_code', IntegerType(), True)
]))
])

# fix data here?

df2 = spark.createDataFrame(data=data, schema=schema2)
# error here:  ValueError: field name_struct: Length of object (2) does not match with length of fields (3)
df2.printSchema()
df2.show()

Спасибо!

Подробнее здесь: https://stackoverflow.com/questions/644 ... ing-new-sc