Gunakan umpan data perubahan pada Azure Databricks

Umpan data perubahan (CDF) melacak perubahan tingkat baris antara versi pada tabel Delta Lake atau tabel Apache Iceberg v3.

Azure Databricks mendukung dua pendekatan:

  • Umpan data perubahan otomatis: Menghitung perubahan selama pembacaan tabel menggunakan metadata silsilah baris. Ini tidak memerlukan konfigurasi tabel individual dan berfungsi pada tabel Delta Lake dan Apache Iceberg v3. Lihat umpan data perubahan otomatis.
  • Umpan data perubahan warisan: Mewujudkan perubahan selama penulisan tabel. Hanya mendukung tabel Delta Lake. Memerlukan konfigurasi tabel individual. Lihat Umpan data perubahan warisan untuk Delta Lake.

Anda dapat menggunakan umpan data perubahan untuk kasus penggunaan data umum termasuk:

  • Pipeline ETL inkremental yang hanya memproses baris data yang berubah sejak pipeline terakhir dijalankan.
  • Jejak audit yang melacak modifikasi data untuk persyaratan kepatuhan dan tata kelola.
  • Beban kerja replikasi data yang menyinkronkan perubahan pada tabel hilir, cache, atau sistem eksternal.

Umpan data perubahan otomatis

Important

Fitur ini ada di Pratinjau Umum. Admin ruang kerja dapat mengontrol akses ke fitur ini dari halaman Pratinjau . Lihat Kelola Pratinjau Azure Databricks.

Umpan data perubahan otomatis menghitung perubahan tingkat baris pada waktu kueri, bukan pada waktu tulis, menggunakan pelacakan baris untuk Delta Lake dan silsilah baris untuk Apache Iceberg v3. Tidak seperti umpan data perubahan warisan, umpan data perubahan otomatis tidak memerlukan konfigurasi tabel individual dan berfungsi pada tabel Delta Lake dan tabel Apache Iceberg v3.

Karena perubahan tidak dihitung pada setiap operasi penulisan untuk operasi MERGE INTO dan UPDATE, umpan data perubahan otomatis meningkatkan kinerja penulisan dan mengurangi biaya penyimpanan dibandingkan dengan umpan data perubahan lama.

Umpan data perubahan otomatis menggunakan API table_changes() dan readChangeFeed yang sama seperti umpan data perubahan versi lama dan kompatibel dengan kueri batch, Structured Streaming, serta Databricks-to-Databricks Delta Lake Sharing. Lihat Membaca perubahan dalam kueri batch dan Memproses perubahan data secara bertahap.

Persyaratan

  • Databricks Runtime 18 atau lebih tinggi
  • Format tabel yang didukung yang terdaftar di Katalog Unity:
    • Tabel terkelola dalam format Delta Lake dengan pelacakan baris diaktifkan atau dalam format Iceberg v3.
    • Tabel eksternal dalam format Delta Lake dengan pelacakan baris diaktifkan.

Lihat Jenis tabel Databricks Unity Catalog.

Note

Umpan data perubahan bukan bagian dari spesifikasi Apache Iceberg. pembaca Azure Databricks dapat meminta umpan data perubahan otomatis untuk tabel Apache Iceberg v3, tetapi pembaca Iceberg eksternal tidak dapat. Lihat spesifikasi tabel Iceberg.

Untuk Delta Lake, hanya pembaca Azure Databricks yang dapat mengkueri umpan data perubahan otomatis.

Gunakan umpan perubahan data

Untuk menggunakan umpan data perubahan, verifikasi bahwa Anda menggunakan tabel yang memenuhi persyaratan. Lihat Persyaratan.

Untuk membaca umpan data perubahan secara batch, lakukan hal berikut:

Python

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("<table_name>")

Scala

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("<table_name>")

SQL

SELECT * FROM table_changes('<table_name>', 0)

Untuk informasi selengkapnya tentang pembacaan batch untuk mengubah umpan data, lihat Membaca perubahan dalam kueri batch.

Untuk membaca umpan data perubahan secara streaming, lakukan hal berikut:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")

Untuk informasi selengkapnya tentang pembacaan streaming untuk mengubah umpan data, lihat Data perubahan proses bertahap.

Migrasikan dari feed data perubahan lama

Untuk memigrasikan tabel Delta Lake dari umpan data perubahan warisan ke umpan data perubahan otomatis, lakukan hal berikut:

  1. Verifikasi bahwa tabel Anda memenuhi persyaratan.
  2. Nonaktifkan umpan data perubahan warisan dengan menjalankan perintah berikut:
ALTER TABLE <table_name> UNSET TBLPROPERTIES ('delta.enableChangeDataFeed');

Anda tidak dapat menggunakan feed data perubahan versi lama dan otomatis secara bersamaan.

Mengubah skema umpan data

Saat Anda membaca dari umpan data perubahan untuk tabel tersebut, kueri menggunakan skema untuk versi tabel terbaru. Azure Databricks mendukung sebagian besar operasi perubahan skema dan evolusi, tetapi tabel dengan pemetaan kolom memiliki batasan. Lihat Tabel dengan pemetaan kolom.

Selain kolom data dari skema tabel Delta Lake, feed data perubahan juga berisi kolom metadata yang mengidentifikasi jenis peristiwa perubahan:

Nama kolom Type Values
_change_type String Berisi: insert, update_preimage, update_postimage, delete.
preimage adalah nilai sebelum pembaruan, postimage adalah nilai setelah pembaruan.
_commit_version Long Berisi: log Delta atau versi tabel yang berisi perubahan.
_commit_timestamp Timestamp Berisi: stempel waktu yang terkait saat commit dibuat.

Jika skema berisi kolom dengan nama yang sama dengan kolom metadata ini, Anda tidak dapat menggunakan umpan data perubahan pada tabel. Sebelum Anda mengaktifkan ubah umpan data, ganti nama kolom di tabel Anda untuk mengatasi konflik ini.

Memproses perubahan data secara bertahap

Databricks merekomendasikan agar Anda menggunakan umpan data perubahan dalam kombinasi dengan Streaming Terstruktur untuk memproses perubahan secara bertahap dari tabel. Anda harus menggunakan Streaming Terstruktur untuk Azure Databricks untuk melacak versi secara otomatis untuk umpan data perubahan tabel Anda. Untuk pemrosesan CDC dengan tabel SCD tipe 1 atau tipe 2, lihat API CDC OTOMATIS: Menyederhanakan perubahan pengambilan data dengan alur.

Saat aliran pertama kali dimulai, feed data perubahan mengembalikan snapshot terbaru dari tabel sebagai record INSERT, lalu mengembalikan perubahan berikutnya sebagai data perubahan. Mengubah umpan data menerapkan perubahan data dan baris data baru ke log transaksi tabel secara bersamaan.

Untuk mengonfigurasi aliran untuk membaca umpan data perubahan tabel, atur opsi readChangeFeed menjadi true sebagai berikut:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")

Pembatasan laju

Azure Databricks mendukung batas laju (maxFilesPerTrigger, maxBytesPerTrigger) dan excludeRegex saat membaca data perubahan. Untuk daftar lengkap opsi streaming Delta Lake, lihat Delta Lake.

Secara opsional, Anda dapat menentukan versi awal, lihat Menentukan versi awal. Untuk versi selain snapshot awal, batas laju diterapkan secara atomik pada seluruh commit. Entah batch saat ini mencakup seluruh commit, atau batch saat ini menunda commit ke batch berikutnya.

Riwayat putar ulang tabel

Umpan data perubahan tidak dimaksudkan untuk berfungsi sebagai rekaman permanen dari semua perubahan pada tabel. Ini hanya mencatat perubahan yang terjadi setelah umpan data perubahan diaktifkan. Anda dapat memulai pembacaan streaming baru untuk menangkap versi saat ini dan semua perubahan berikutnya.

Rekaman dalam umpan data perubahan bersifat sementara dan hanya dapat diakses untuk jendela retensi tertentu. Log transaksi menghapus versi tabel dan versi umpan data perubahan yang sesuai secara berkala. Saat versi dihapus, Anda tidak dapat lagi membaca umpan data perubahan untuk versi tersebut.

Mengarsipkan data perubahan untuk riwayat permanen

Jika kasus penggunaan mengharuskan Anda mempertahankan riwayat permanen semua perubahan pada tabel, gunakan logika bertahap untuk menulis rekaman dari umpan data perubahan ke tabel baru.

Contoh berikut menunjukkan penggunaan trigger.AvailableNow untuk memproses data yang tersedia sebagai beban kerja batch untuk mengaudit atau memutar ulang perubahan penuh:

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)
Scala
spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("target_table")

Tentukan versi awal

Untuk membaca perubahan dari titik tertentu, tentukan versi awal menggunakan tanda waktu atau nomor versi. Versi awal diperlukan untuk pembacaan batch. Secara opsional, Anda dapat menentukan versi akhir untuk membatasi rentang. Untuk mempelajari selengkapnya tentang riwayat tabel, lihat Perjalanan waktu.

Saat Anda mengonfigurasi beban kerja Streaming Terstruktur yang menggunakan umpan data perubahan, menentukan versi awal dapat memengaruhi performa pemrosesan:

  • Alur pemrosesan data baru biasanya mendapat manfaat dari perilaku default, yang merekam semua rekaman yang ada dalam tabel sebagai INSERT operasi saat aliran pertama kali dimulai.
  • Jika tabel target Anda sudah berisi semua rekaman dengan perubahan yang sesuai hingga titik tertentu, tentukan versi awal untuk menghindari pemrosesan status tabel sumber sebagai INSERT peristiwa.

Contoh berikut menunjukkan cara pulih dari kegagalan streaming dengan checkpoint yang rusak. Dalam contoh ini, asumsikan kondisi berikut:

  1. Umpan data perubahan diaktifkan pada tabel sumber pada pembuatan tabel.
  2. Tabel hilir target memproses semua perubahan hingga dan termasuk versi 75.
  3. Riwayat versi untuk tabel sumber tersedia untuk versi 70 ke atas.

Saat Anda menentukan aliran tulis ke tabel target yang ada, Anda harus menentukan lokasi titik pemeriksaan baru:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")

Important

Jika Anda menentukan versi awal dan versi tersebut tidak tersedia dalam riwayat tabel, aliran gagal dimulai dari titik pemeriksaan baru. Karena tabel terkelola membersihkan versi historis secara otomatis, semua versi awal yang ditentukan akhirnya dihapus.

Lihat Memutar ulang riwayat tabel.

Membaca perubahan dalam kueri berkelompok

Anda dapat menggunakan sintaks kueri batch untuk membaca semua perubahan mulai dari versi tertentu atau membaca perubahan dalam rentang versi tertentu sebagai berikut:

  • Tentukan versi sebagai bilangan bulat dan tanda waktu sebagai string dalam format yyyy-MM-dd[ HH:mm:ss[.SSS]].
  • Versi awal dan akhir bersifat inklusif. Untuk membaca dari versi awal ke versi terbaru, tentukan hanya versi awal.
  • Jika Anda menentukan versi sebelum umpan data perubahan diaktifkan, itu menimbulkan kesalahan.

Untuk menggunakan pembacaan batch dengan opsi versi awal dan akhir, lakukan hal berikut:

SQL

Untuk membaca dari versi 0 ke 10, lakukan hal berikut:

SELECT * FROM table_changes('tableName', 0, 10)

Untuk membaca antara dua versi tanda waktu, lakukan hal berikut:

--
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

Untuk membaca dari versi awal ke versi terbaru, lakukan hal berikut:

SELECT * FROM table_changes('tableName', 0)

Untuk membaca perubahan untuk tabel dengan karakter khusus dalam nama, lakukan hal berikut:

SELECT * FROM table_changes('`schema`.`dotted.tableName`', '2021-04-21 06:45:46', '2021-05-21 12:00:00')

Lihat fungsi bernilai tabel table_changes.

Python

Untuk membaca dari versi 0 ke 10, lakukan hal berikut:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .option("endingVersion", 10) \
  .table("myDeltaTable")

Untuk membaca di antara dua tanda waktu, lakukan hal berikut:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingTimestamp", '2021-04-21 05:45:46') \
  .option("endingTimestamp", '2021-05-21 12:00:00') \
  .table("myDeltaTable")

Untuk membaca dari versi awal ke versi terbaru, lakukan hal berikut:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

Scala

Untuk membaca dari versi 0 ke 10, lakukan hal berikut:

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .option("endingVersion", 10)
  .table("myDeltaTable")

Untuk membaca di antara dua tanda waktu, lakukan hal berikut:

spark.read
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .option("endingTimestamp", "2021-05-21 12:00:00")
  .table("myDeltaTable")

Untuk membaca dari versi awal ke versi terbaru, lakukan hal berikut:

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

Menangani versi di luar rentang

Secara default, jika Anda menentukan versi atau tanda waktu yang melebihi penerapan terakhir, kueri mengembalikan kesalahan timestampGreaterThanLatestCommit.

Di Databricks Runtime 11.3 LTS ke atas, Anda dapat mengaktifkan toleransi untuk versi di luar rentang sebagai berikut:

SET spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;

Saat konfigurasi ini diaktifkan, kueri mengembalikan hasil yang berbeda sebagai berikut:

  • Versi atau stempel waktu awal yang melewati commit terakhir akan mengembalikan hasil kosong.
  • Versi akhir atau tanda waktu di luar penerapan terakhir mengembalikan semua perubahan dari awal hingga penerapan terakhir.

Umpan data perubahan warisan untuk Delta Lake

Umpan data perubahan warisan memerlukan konfigurasi manual untuk tabel Delta Lake individual. Karena umpan data perubahan tidak disertakan dalam spesifikasi Apache Iceberg, tabel Apache Iceberg tidak didukung. Databricks merekomendasikan agar Anda bermigrasi ke umpan data perubahan otomatis. Lihat Bermigrasi dari feed data perubahan lama.

Saat change data feed lama diaktifkan, runtime mencatat peristiwa perubahan untuk semua data yang ditulis ke tabel. Ini termasuk data baris bersama dengan metadata yang menunjukkan apakah baris yang ditentukan disisipkan, dihapus, atau diperbarui.

Umpan data perubahan lawas menggunakan API pembacaan readChangeFeed dan table_changes() yang sama seperti umpan data perubahan otomatis. Lihat Memproses perubahan data secara bertahap dan Membaca perubahan dalam kueri batch.

Mengaktifkan umpan data perubahan warisan

Anda harus secara eksplisit mengaktifkan umpan data perubahan warisan pada tabel individual. Pilih salah satu metode berikut:

Tabel baru

Atur properti delta.enableChangeDataFeed = true tabel dalam CREATE TABLE perintah .

CREATE TABLE student (id INT, name STRING, age INT)
  TBLPROPERTIES (delta.enableChangeDataFeed = true)

Note

Jika Anda menonaktifkan umpan data perubahan warisan untuk interval waktu apa pun lalu mengaktifkannya lagi, interval tidak akan dapat dikueri. Gunakan umpan data perubahan otomatis untuk mengkueri perubahan selama interval. Lihat umpan data perubahan otomatis.

Tabel yang sudah ada

Atur properti delta.enableChangeDataFeed = true tabel dalam ALTER TABLE perintah .

ALTER TABLE myDeltaTable
  SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

Pertimbangan penyimpanan

Tabel terkelola merekam perubahan data secara efisien dan mungkin menggunakan fitur lain untuk mengoptimalkan tata letak penyimpanan.

Dengan umpan data perubahan warisan, Anda harus mempertimbangkan perilaku penyimpanan berikut:

  • Anda mungkin melihat peningkatan kecil biaya penyimpanan karena perubahan mungkin dicatat dalam file terpisah.
  • Beberapa operasi, seperti operasi insert-only atau penghapusan seluruh partisi, tidak menghasilkan file data perubahan. Azure Databricks menghitung umpan data perubahan langsung dari log transaksi.
  • Ubah file data menggunakan kebijakan penyimpanan tabel. Perintah VACUUM menghapus file data perubahan, dan perubahan dari kebijakan penyimpanan titik pemeriksaan penggunaan log transaksi.

Databricks menyarankan agar Anda tidak mencoba merekonstruksi umpan data perubahan dengan langsung mengkueri file data perubahan. Selalu gunakan API Delta Lake dan Apache Iceberg.

Keterbatasan

Pertimbangkan batasan berikut untuk mengubah umpan data:

Tabel dengan pemetaan kolom

Dengan pemetaan kolom diaktifkan pada tabel Delta Lake, Anda dapat menghapus atau mengganti nama kolom tanpa menulis ulang file data. Lihat mengganti nama dan menghapus kolom dengan menggunakan pemetaan kolom Delta Lake.

Namun, umpan data perubahan memiliki batasan setelah perubahan skema non-aditif. Perubahan skema non-aditif mencakup operasi berikut:

  • Ubah nama atau hapus kolom.
  • Ubah jenis data kolom.
  • Ubah kemampuan kolom untuk menerima nilai null, misalnya dengan ALTER COLUMN ... SET NOT NULL. Lihat NOT NULL batasan.

Anda tidak dapat membaca feed data perubahan untuk transaksi atau rentang tempat perubahan skema non-aditif terjadi.

Untuk mengakomodasi perubahan skema non-aditif sebelum atau sesudah rentang pembacaan batch yang ditentukan, kueri menggunakan skema dari versi akhir rentang tersebut, bukan versi tabel terbaru. Kueri masih gagal jika rentang versi mencakup perubahan skema yang tidak aditif.

Umpan data perubahan secara otomatis

  • Karena umpan data perubahan tidak didukung dalam spesifikasi Apache Iceberg, klien Iceberg eksternal tidak dapat mengkueri umpan data perubahan otomatis. Lihat spesifikasi tabel Iceberg.
  • Untuk transaksi multi-pernyataan, jika tabel sumber dimodifikasi selama transaksi, umpan data perubahan otomatis tidak didukung.
  • Umpan data perubahan otomatis tidak didukung pada tabel dengan filter baris atau masker kolom. Lihat Filter baris dan masker kolom.
  • Kueri umpan data perubahan tidak dapat mencakup versi tabel tempat perubahan skema non-aditif terjadi, seperti perubahan nama kolom, penurunan, atau perubahan tipe data. Pisahkan kueri menjadi rentang sebelum dan sesudah perubahan skema.
  • Last updated on