Вставить или обновить дельта-таблицу из фрейма данных в Pyspark

В настоящее время у меня есть фрейм данных pyspark, из которого я изначально создал дельта-таблицу, используя приведенный ниже код -

df.write.format("delta").saveAsTable("events")

Теперь, поскольку приведенный выше фрейм данных заполняет данные ежедневно по моему требованию, следовательно, для добавления новых записей в дельта-таблицу я использовал синтаксис ниже -

df.write.format("delta").mode("append").saveAsTable("events")

Теперь все это я сделал в блоках данных и в своем кластере. Я хочу знать, как я могу написать общий код pyspark на python, который будет создавать дельта-таблицу, если она не существует, и добавлять записи, если дельта-таблица существует. Я хочу это сделать, потому что, если я передам кому-то свой пакет python, они не будут имеют одинаковую дельта-таблицу в своей среде, поэтому она должна создаваться динамически из кода.


apache-spark pyspark delta-lake
person Tushaar    schedule 23.02.2021    source источник
comment
Это в точности определение режима добавления при записи.   -  person blackbishop    schedule 24.02.2021

Ответы (1)


arrow_upward
0
arrow_downward

Если у вас еще нет таблицы Delta, она будет создана при использовании режима append. Таким образом, вам не нужно писать какой-либо специальный код для обработки случая, когда таблица еще не существует и когда она завершается.

P.S. Такой код понадобится вам только в том случае, если вы выполняете слияние в таблицу, а не добавление. В этом случае код будет выглядеть так:

if table_exists:
  do_merge
else:
  df.write....
person Alex Ott    schedule 27.02.2021