Menggantikan data secara selektif dengan Delta Lake

Delta Lake memiliki opsi berbeda berikut untuk penimpaan selektif:

Option Skenario penggunaan Jenis komputasi yang didukung Versi minimum
REPLACE WHERE Timpa baris secara atomik yang sesuai dengan predikat. Gunakan untuk penggantian dengan kondisi pencocokan tetap, seperti colA = 5 atau int_col IN (1, 2, 3). Semua jenis komputasi. SQL di Databricks Runtime 12.2 LTS ke atas. Python dan Scala di Databricks Runtime 9.1 LTS ke atas.
REPLACE USING Penimpaan data dinamis. Mengganti semua baris yang cocok dengan kolom yang ditentukan, berdasarkan perbandingan kesetaraan nilai kolom, dalam himpunan data yang disediakan. Semua jenis komputasi. SQL di Databricks Runtime 16.3 ke atas. Python dan Scala di Databricks Runtime 18.2 ke atas.
REPLACE ON Data dinamis ditimpa menggunakan ekspresi boolean. Gunakan untuk penggantian dengan kondisi pencocokan yang kompleks atau aman NULL, seperti s.colA <=> t.colA AND s.colB <=> t.colB. Semua jenis komputasi. SQL di Databricks Runtime 17.1 ke atas. Python dan Scala di Databricks Runtime 18.2 ke atas.
partitionOverwriteMode Penimpaan partisi dinamis lama, yang menimpa semua data yang ada di setiap partisi tempat penulisan akan mengkomitkan data baru. Tidak disarankan untuk beban kerja baru. SQL hanya mendukung komputasi klasik. Python dan Scala mendukung semua jenis komputasi. SQL, Python, dan Scala di Databricks Runtime 11.3 LTS ke atas.

Untuk sebagian besar kasus penggunaan, Databricks merekomendasikan penggunaan REPLACE USING atau REPLACE WHERE. Gunakan REPLACE ON hanya jika skenario penggunaan Anda memerlukan kondisi pencocokan yang kompleks atau yang aman terhadap NULL.

Untuk detail tentang perilaku penggantian setiap opsi, lihat INSERT. Untuk daftar DataFrameWriter lengkap opsi Delta Lake, lihat Delta Lake dan Apache Iceberg.

Di Scala dan Python, replaceOn dan replaceUsing tidak dapat digunakan dalam kombinasi dengan replaceWhere, partitionOverwriteMode, atau overwriteSchema.

Untuk kueri sumber kosong, baik REPLACE USING maupun REPLACE ON tidak menghapus data, namun REPLACE WHERE mungkin menghapus data.

Important

Jika data ditimpa secara tidak sengaja, Anda dapat menggunakan mengembalikan untuk membatalkan ubahannya.

REPLACE WHERE

Anda dapat secara selektif menimpa hanya data yang cocok dengan ekspresi arbitrer dengan REPLACE WHERE.

Important

Untuk mendapatkan manfaat dari refresh bertahap saat menjalankan REPLACE WHERE, gunakan alur REPLACE WHERE di Spark Declarative Pipelines (SDP). Lihat Pemrosesan batch dengan alur REPLACE WHERE.

Untuk mengganti event di bulan Januari secara atomik dalam tabel target, yang dipartisi berdasarkan start_date, dengan data dalam replace_data:

Python

(replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")
)

Scala

replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events REPLACE WHERE start_date >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data

Kode sampel ini menulis data dalam replace_data, memvalidasi bahwa semua baris cocok dengan predikat, dan melakukan penggantian atom menggunakan overwrite semantik. Jika ada nilai dalam operasi yang berada di luar predikat, operasi ini gagal dengan kesalahan secara default.

Pada komputasi klasik, untuk mengubah perilaku ini menjadi overwrite nilai dalam rentang predikat dan insert rekaman di luar rentang yang ditentukan, hapus pemeriksaan batasan dengan mengatur spark.databricks.delta.replaceWhere.constraintCheck.enabled ke false:

Python

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false

Note

REPLACE WHERE menerima sebuah boolean_expression dengan beberapa batasan. Lihat INSERT di referensi bahasa SQL.

Untuk kueri sumber kosong, REPLACE WHERE mungkin menghapus baris tabel.

Perilaku warisan

Legacy replaceWhere hanya tersedia pada compute klasik. Lihat Gambaran umum komputasi klasik.

Jika Anda menggunakan perilaku lama replaceWhere, kueri hanya menimpa data di kolom partisi yang cocok dengan predikat. Perintah berikut akan secara atomik menggantikan bulan Januari dalam tabel target, yang dipartisi oleh date, dengan data dalam df:

Python
(df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")
)
Scala
df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")

Untuk menggunakan perilaku warisan, atur spark.databricks.delta.replaceWhere.dataColumns.enabled ke false:

Python
spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)
Scala
spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)
SQL
SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false

Penimpaan data dinamis

Penimpaan data dinamis secara selektif menggantikan data yang sesuai dengan kolom kunci atau ekspresi Boolean yang ditentukan, sementara semua data lainnya tetap tidak berubah. Tabel yang dipartisi, tabel yang tidak dipartisi, dan tabel dengan pengklusteran cair semuanya didukung.

Penimpaan partisi dinamis merupakan bagian dari perilaku penimpaan data dinamis. Penimpaan partisi dinamis menggantikan semua data yang ada di setiap partisi tempat operasi penulisan akan mengomiten data baru dan membiarkan semua partisi lainnya tetap tidak berubah. Hanya tabel yang dipartisi yang didukung.

REPLACE USING

SQL didukung di Databricks Runtime 16.3 ke atas. Python dan Scala didukung di Databricks Runtime 18.2 ke atas. Untuk perbedaan perilaku dalam Databricks Runtime 16.3 hingga 17.1 lihat Perilaku warisan.

REPLACE USING memungkinkan perilaku timpa atomik yang independen dari komputasi, yang berfungsi pada gudang Databricks SQL, komputasi tanpa server, dan komputasi klasik. REPLACE USING tidak mengharuskan Anda mengatur konfigurasi sesi Spark.

REPLACE USING menggantikan baris ketika kolom yang ditentukan dianggap sama berdasarkan operator kesetaraan. Semua data lainnya tetap tidak berubah.

Gunakan penimpaan data dinamis dengan REPLACE USING:

Python

(sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events
  REPLACE USING (event_id, start_date)
  SELECT * FROM source_data

Untuk kueri sumber kosong, REPLACE USING tidak menghapus baris tabel apa pun.

Untuk kondisi pencocokan yang kompleks atau aman NULL, gunakan REPLACE ON sebagai gantinya. Lihat REPLACE ON.

Lihat INSERT di referensi bahasa SQL.

Perilaku warisan

Dalam Databricks Runtime 16.3 hingga 17.1, REPLACE USING menggunakan perilaku warisan dan hanya memungkinkan penimpaan partisi dinamis, sedangkan Databricks Runtime 17.2 ke atas memungkinkan penimpaan data dinamis.

Perhatikan batasan dan perilaku berikut terkait perilaku lama REPLACE USING:

  • Anda harus menentukan kumpulan lengkap kolom partisi tabel dalam USING klausa.
  • Selalu validasi bahwa data yang ditulis hanya menyentuh partisi yang diharapkan. Baris tunggal di partisi yang salah dapat secara tidak sengaja menimpa seluruh partisi.

REPLACE ON

SQL didukung di Databricks Runtime 17.1 ke atas. Python dan Scala didukung di Databricks Runtime 18.2 ke atas.

REPLACE ON menggantikan baris saat cocok dengan kondisi yang ditentukan pengguna, tidak seperti REPLACE USING, yang menggantikan baris saat kolom yang ditentukan dibandingkan sama di bawah kesetaraan. Gunakan REPLACE ON saat Anda memerlukan logika pencocokan yang REPLACE USING tidak mendukung, seperti memperlakukan NULL nilai sebagai sama.

Secara opsional, gunakan targetAlias opsi untuk menentukan alias untuk tabel target dan .as() atau .alias() API untuk menentukan alias untuk data sumber.

Untuk sintaks SQL, lihat INSERT.

Python

(sourceDataDF.alias("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.as("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events AS t
  REPLACE ON (s.event_id <=> t.event_id AND s.start_date <=> t.start_date)
  (SELECT * FROM source_data) AS s

Untuk kueri sumber kosong, REPLACE ON tidak menghapus baris tabel apa pun.

Partisi dinamis ditimpa dengan partitionOverwriteMode (warisan)

Important

Fitur ini ada di Pratinjau Publik.

Databricks Runtime 11.3 LTS ke atas mendukung penimpaan partisi dinamis untuk tabel yang dipartisi menggunakan mode timpa: entah INSERT OVERWRITE dalam SQL atau penulisan DataFrame dengan df.write.mode("overwrite"). Jenis penimpaan ini hanya tersedia untuk komputasi klasik, bukan gudang Databricks SQL atau komputasi tanpa server.

Peringatan

Jika memungkinkan, gunakan INSERT REPLACE USING alih-alih menimpa partisi dengan INSERT OVERWRITE PARTITION dan spark.sql.sources.partitionOverwriteMode=dynamic. Penggantian partisi dapat menggunakan data usang ketika perubahan pemartisian terjadi.

Untuk menggunakan mode timpa partisi dinamis, atur konfigurasi spark.sql.sources.partitionOverwriteMode sesi Spark ke dynamic. Atau, Anda dapat mengatur opsi DataFrameWriter ke partitionOverwriteModedynamic. Jika ada, opsi khusus kueri menggantikan mode yang ditentukan dalam konfigurasi sesi. Nilai defaultnya untuk spark.sql.sources.partitionOverwriteMode adalah static.

Contoh berikut menggunakan partitionOverwriteMode:

SQL

SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;

Python

(df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")

Ingatlah batasan dan perilaku berikut untuk partitionOverwriteMode:

  • Anda tidak dapat mengatur overwriteSchema ke true.
  • Anda tidak dapat menentukan partitionOverwriteMode dan replaceWhere secara bersamaan dalam operasi DataFrameWriter yang sama.
  • Jika Anda menentukan replaceWhere kondisi menggunakan DataFrameWriter opsi, Delta Lake menerapkan kondisi tersebut untuk mengontrol data mana yang ditimpa. Opsi ini lebih diutamakan daripada partitionOverwriteMode konfigurasi tingkat sesi.
  • Selalu validasi bahwa data yang ditulis hanya menyentuh partisi yang diharapkan. Baris tunggal di partisi yang salah dapat secara tidak sengaja menimpa seluruh partisi.
  • Last updated on