Обновление таблицы с помощью PySpark в Azure Databricks. ⇐ Python
-
Anonymous
Обновление таблицы с помощью PySpark в Azure Databricks.
У меня есть таблица в базе данных SQL Server
создать таблицу person (Имя varchar(255), Фамилия varchar(255)) И я пытаюсь выполнить простую операцию обновления с помощью PySpark:
# Читаем данные из таблицы «person» person_df = spark.read.jdbc(url=database_url, table="person", Properties=properties) # Получить все записи из таблицы «person» person_df.show() # Проверяем, существует ли уже человек с именем «Джек» существующие_записи = person_df.filter(person_df["Имя"] == "Джек") если существующие_записи.count() == 0: # Человек с именем «Джек» не существует, поэтому добавьте его в DataFrame печать("Не найден") new_person = [("Джек", "Браун")] new_person_df = spark.createDataFrame(new_person, ["Имя", "Фамилия"]) обновленный_person_df = person_df.union(new_person_df) еще: печать("Найдено") # Человек с именем «Джек» уже существует, поэтому обновите его фамилию на «Браун». обновленный_person_df = person_df.withColumn("Фамилия", F.when(person_df["Имя"] == "Джек", "Белый") .иначе(person_df["Фамилия"])) # Показать обновленные данные обновленный_person_df.show() # Сохраняем обновленные данные обратно в таблицу «person». обновленный_person_df.write.jdbc(url=database_url, таблица="человек", свойства = свойства, режим = «перезаписать») Это хорошо работает, если в таблице нет записи. Но после вставки записи, даже если кадр данных update_person_df верен, эта операция удаляет все записи в таблице.
У меня есть таблица в базе данных SQL Server
создать таблицу person (Имя varchar(255), Фамилия varchar(255)) И я пытаюсь выполнить простую операцию обновления с помощью PySpark:
# Читаем данные из таблицы «person» person_df = spark.read.jdbc(url=database_url, table="person", Properties=properties) # Получить все записи из таблицы «person» person_df.show() # Проверяем, существует ли уже человек с именем «Джек» существующие_записи = person_df.filter(person_df["Имя"] == "Джек") если существующие_записи.count() == 0: # Человек с именем «Джек» не существует, поэтому добавьте его в DataFrame печать("Не найден") new_person = [("Джек", "Браун")] new_person_df = spark.createDataFrame(new_person, ["Имя", "Фамилия"]) обновленный_person_df = person_df.union(new_person_df) еще: печать("Найдено") # Человек с именем «Джек» уже существует, поэтому обновите его фамилию на «Браун». обновленный_person_df = person_df.withColumn("Фамилия", F.when(person_df["Имя"] == "Джек", "Белый") .иначе(person_df["Фамилия"])) # Показать обновленные данные обновленный_person_df.show() # Сохраняем обновленные данные обратно в таблицу «person». обновленный_person_df.write.jdbc(url=database_url, таблица="человек", свойства = свойства, режим = «перезаписать») Это хорошо работает, если в таблице нет записи. Но после вставки записи, даже если кадр данных update_person_df верен, эта операция удаляет все записи в таблице.
