Почему новые столбцы, добавленные в паркетные столы, недоступны из заданий ETL клеевого pyspark?

Question

Мы изучали использование Glue для преобразования некоторых данных JSON в паркет.Один из сценариев, который мы попробовали, - добавление столбца в паркетный столТаким образом, раздел 1 имеет столбцы [A], а раздел 2 имеет столбцы [A, B].Затем мы хотели написать дополнительные задания Glue ETL для агрегирования таблицы паркета, но новый столбец был недоступен.Используя glue_context.create_dynamic_frame.from_catalog для загрузки динамического кадра, наш новый столбец никогда не входил в схему.

Мы попробовали несколько конфигураций для нашего сканера таблиц.Использование единой схемы для всех разделов, единой схемы для пути s3, схемы для каждого раздела.Мы всегда могли видеть новый столбец в данных таблицы Glue, но он всегда был нулевым, если мы запрашивали его из задания Glue с помощью pyspark.Колонка была в паркете, когда мы загрузили некоторые образцы и стали доступны для запросов через Athena.

Почему новые колонки недоступны для pyspark?

Answer 1

Это оказалось проблемой конфигурации искры.Начиная с spark docs :

Как и Protocol Buffer, Avro и Thrift, Parquet также поддерживает эволюцию схемы.Пользователи могут начать с простой схемы и постепенно добавлять дополнительные столбцы в схему по мере необходимости.Таким образом, пользователи могут получить несколько файлов Parquet с разными, но взаимно совместимыми схемами.Источник данных Parquet теперь может автоматически обнаруживать этот случай и объединять схемы всех этих файлов.

Поскольку объединение схем является относительно дорогой операцией и в большинстве случаев не является необходимостью, мы отключили ее по умолчанию.начиная с 1.5.0.Вы можете включить его,

1. , установив для параметра источника данных mergeSchema значение true при чтении файлов Parquet (как показано в примерах ниже), или
2. , установив глобальный параметр SQL spark.sql.parquet.mergeSchema в true.

Мы можем включить объединение схем двумя способами.

1. установить параметр в сеансе spark spark.conf.set("spark.sql.parquet.mergeSchema", "true")
2. установите mergeSchema в true в additional_options при загрузке динамического кадра.

source = glueContext.create_dynamic_frame.from_catalog(
   database="db",
   table_name="table",
   additional_options={"mergeSchema": "true"}
)

После этого новый столбец стал доступен в схеме фрейма.