Как разбить и записать DataFrame в Spark без удаления разделов без новых данных?

Я пытаюсь сохранить DataFrame в HDFS в формате Паркета, используя DataFrameWriter, разделенный на три значения столбца, например:

dataFrame.write.mode(SaveMode.Overwrite).partitionBy("eventdate", "hour", "processtime").parquet(path)

Как уже упоминалось в этом вопросе, partitionBy удалит всю существующую иерархию разделов в path и заменит их на разделы в DataFrame. Поскольку новые инкрементные данные для определенного дня будут поступать периодически, я хочу заменить только те разделы в иерархии, для которых DataFrame имеет данные, а остальные остаются нетронутыми.

Для этого мне нужно сохранить каждый раздел отдельно, используя его полный путь, примерно так:

singlePartition.write.mode(SaveMode.Overwrite).parquet(path + "/eventdate=2017-01-01/hour=0/processtime=1234567890")

Однако у меня возникли проблемы с пониманием лучшего способа организации данных в однораздельный DataFrame, чтобы я мог писать их с помощью их полного пути. Одна идея была примерно такой:

dataFrame.repartition("eventdate", "hour", "processtime").foreachPartition ...

Но foreachPartition работает с Iterator[Row], который не идеален для записи в формат Паркета.

Я также рассмотрел использование select...distinct eventdate, hour, processtime для получения списка разделов, а затем фильтрацию исходного фрейма данных каждым из этих разделов и сохранение результатов до их полного секционированного пути. Но отдельный запрос плюс фильтр для каждого раздела не кажется очень эффективным, так как это будет много операций фильтрации/записи.

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

Спасибо за чтение.

Искра версия: 2.1

Ответ 1

У опции режима Append есть улов!

df.write.partitionBy("y","m","d")
.mode(SaveMode.Append)
.parquet("/data/hive/warehouse/mydbname.db/" + tableName)

Я протестировал и увидел, что это сохранит существующие файлы разделов. Однако на этот раз проблема следующая: если вы дважды запустите тот же код (с теми же данными), он создаст новые паркетные файлы вместо замены существующих для тех же данных (Spark 1.6). Поэтому вместо Append мы можем решить эту проблему с помощью Overwrite. Вместо перезаписи на уровне таблицы мы должны перезаписать на уровне раздела.

df.write.mode(SaveMode.Overwrite)
.parquet("/data/hive/warehouse/mydbname.db/" + tableName + "/y=" + year + "/m=" + month + "/d=" + day)

Для получения дополнительной информации см. следующую ссылку:

Перезаписать определенные разделы в методе записи данных в режиме искробезопасности

(Я обновил свой ответ после комментария сурианто. Thnx.)

Ответ 2

Я знаю, что это очень старый. Поскольку я не вижу опубликованного решения, я отправлю его. Этот подход предполагает, что у вас есть таблица кустов в каталоге, в который вы хотите записать. Один из способов решения этой проблемы - создать временное представление из dataFrame, которое необходимо добавить в таблицу, а затем использовать обычную команду insert overwrite table ..., похожую на куст:

dataFrame.createOrReplaceTempView("temp_view")
spark.sql("insert overwrite table table_name partition ('eventdate', 'hour', 'processtime')select * from temp_view")

Сохраняет старые разделы, перезаписывая только новые разделы.

Ответ 3

Вы можете попробовать режим как добавить.

dataFrame.write.format("parquet")
.mode("append")
.partitionBy("year","month")
.option("path",s"$path/table_name")
.saveAsTable(s"stg_table_name")

Ответ 4

Это старая тема, но у меня возникла та же проблема, и я нашел другое решение, просто установите динамический режим перезаписи раздела, используя:

spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')

Итак, мой сеанс зажигания настроен так:

spark = SparkSession.builder.appName('AppName').getOrCreate()
spark.conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')