API TERAPKAN PERUBAHAN: Menyederhanakan pengambilan data perubahan dengan DLT

DLT menyederhanakan pengambilan data perubahan (CDC) dengan API APPLY CHANGES dan APPLY CHANGES FROM SNAPSHOT. Antarmuka yang Anda gunakan tergantung pada sumber data perubahan:

• Gunakan APPLY CHANGES untuk memproses perubahan dari umpan data perubahan (CDF).
• Gunakan APPLY CHANGES FROM SNAPSHOT (Pratinjau Umum) untuk memproses perubahan rekam jepret database.

Sebelumnya, pernyataan MERGE INTO umumnya digunakan untuk memproses rekaman CDC di Azure Databricks. Namun, MERGE INTO dapat menghasilkan hasil yang salah karena rekaman yang tidak berurutan atau memerlukan logika yang kompleks untuk memesan ulang rekaman.

API APPLY CHANGES didukung di antarmuka DLT SQL dan Python. API APPLY CHANGES FROM SNAPSHOT didukung di antarmuka DLT Python.

Baik APPLY CHANGES maupun APPLY CHANGES FROM SNAPSHOT mendukung pembaruan tabel menggunakan SCD jenis 1 dan tipe 2:

• Gunakan SCD tipe 1 untuk memperbarui rekaman secara langsung. Riwayat tidak dipertahankan untuk rekaman yang diperbarui.
• Gunakan SCD tipe 2 untuk menyimpan riwayat rekaman, baik pada semua pembaruan atau pembaruan untuk sekumpulan kolom tertentu.

Untuk sintaks dan referensi lainnya, lihat:

• Mengubah pengambilan data dari umpan perubahan dengan Python di DLT
• Mengubah pengambilan data dengan SQL di DLT

Persyaratan

Untuk menggunakan CDC API, alur Anda harus dikonfigurasi untuk menggunakan alur DLT tanpa server atau DLT Pro atau edisi Advanced.

Bagaimana CDC diimplementasikan dengan API APPLY CHANGES?

Dengan secara otomatis menangani rekaman di luar urutan, API APPLY CHANGES di DLT memastikan pemrosesan rekaman CDC yang benar dan menghapus kebutuhan untuk mengembangkan logika kompleks untuk menangani rekaman yang tidak berurutan. Anda harus menentukan kolom dalam data sumber untuk mengurutkan rekaman, yang ditafsirkan DLT sebagai representasi yang meningkat secara monoton dari pengurutan data sumber yang tepat. DLT secara otomatis menangani data yang tiba di luar urutan. Untuk perubahan SCD tipe 2, DLT menyebarluaskan nilai urutan yang sesuai ke kolom __START_AT dan __END_AT tabel target. Harus ada satu pembaruan yang berbeda per kunci pada setiap nilai urutan, dan nilai urutan NULL tidak didukung.

Untuk melakukan pemrosesan CDC dengan APPLY CHANGES, Anda terlebih dahulu membuat tabel streaming lalu menggunakan pernyataan APPLY CHANGES INTO di SQL atau fungsi apply_changes() di Python untuk menentukan sumber, kunci, dan urutan umpan perubahan. Untuk membuat tabel streaming target, gunakan pernyataan CREATE OR REFRESH STREAMING TABLE di SQL atau fungsi create_streaming_table() di Python. Lihat contoh SCD tipe 1 dan tipe 2 pemrosesan.

Untuk detail sintaks, lihat referensi DLT SQL atau referensi Python.

Bagaimana CDC diimplementasikan dengan API APPLY CHANGES FROM SNAPSHOT?

APPLY CHANGES FROM SNAPSHOT adalah API deklaratif yang secara efisien menentukan perubahan data sumber dengan membandingkan serangkaian rekam jepret dalam urutan dan kemudian menjalankan pemrosesan yang diperlukan untuk pemrosesan CDC rekaman dalam rekam jepret. APPLY CHANGES FROM SNAPSHOT hanya didukung oleh antarmuka DLT Python.

APPLY CHANGES FROM SNAPSHOT mendukung penyerapan rekam jepret dari beberapa jenis sumber:

• Gunakan penyerapan rekam jepret berkala untuk menyerap rekam jepret dari tabel atau tampilan yang ada. APPLY CHANGES FROM SNAPSHOT memiliki antarmuka yang sederhana dan efisien untuk mendukung penyerapan rekam jepret secara berkala dari objek database yang ada. Sebuah cuplikan baru dimasukkan dengan setiap pembaruan alur kerja, dan waktu pengambilan digunakan sebagai versi cuplikan. Saat alur data dijalankan dalam mode berkelanjutan, beberapa cuplikan diambil dengan setiap pembaruan alur data pada periode yang ditentukan oleh pengaturan interval pemicu untuk alur data yang berisi pemrosesan APPLY CHANGES FROM SNAPSHOT.
• Gunakan penyerapan rekam jepret historis untuk memproses file yang berisi rekam jepret database, seperti rekam jepret yang dihasilkan dari database Oracle atau MySQL atau gudang data.

Untuk melakukan pemrosesan CDC dari jenis sumber apa pun dengan APPLY CHANGES FROM SNAPSHOT, Anda terlebih dahulu membuat tabel streaming lalu menggunakan fungsi apply_changes_from_snapshot() di Python untuk menentukan rekam jepret, kunci, dan argumen lain yang diperlukan untuk mengimplementasikan pemrosesan. Lihat contoh penyerapan rekam jepret berkala dan penyerapan rekam jepret historis.

Rekam jepret yang diteruskan ke API harus dalam urutan naik berdasarkan versi. Jika DLT mendeteksi rekam jepret yang tidak berurutan, kesalahan akan muncul.

Untuk detail sintaks, lihat referensi DLT Python.

Keterbatasan

Kolom yang digunakan untuk pengurutan harus merupakan tipe data yang dapat diurutkan.

Contoh : Pemrosesan SCD tipe 1 dan SCD tipe 2 dengan data sumber CDF

Bagian berikut ini menyediakan contoh kueri DLT SCD tipe 1 dan tipe 2 yang memperbarui tabel target berdasarkan peristiwa sumber dari umpan data perubahan yang:

1. Membuat catatan pengguna baru.
2. Menghapus rekaman pengguna.
3. Memperbarui catatan pengguna. Dalam contoh SCD tipe 1, operasi terakhir UPDATE tiba terlambat dan dihilangkan dari tabel target, menunjukkan penanganan kejadian yang tidak berurutan.

Contoh berikut mengasumsikan keakraban dengan mengonfigurasi dan memperbarui alur DLT. Lihat Tutorial : Jalankan alur DLT pertama Anda.

Untuk menjalankan contoh ini, Anda harus mulai dengan membuat himpunan data sampel. Lihat Hasilkan data pengujian.

Berikut ini adalah catatan input untuk contoh-contoh ini:

Jika Anda membatalkan komentar baris terakhir dalam contoh data, baris tersebut akan menyisipkan catatan berikut yang menentukan di mana kumpulan data harus dipotong.

Pembaruan Proses SCD Tipe 1

Contoh berikut menunjukkan pemrosesan pembaruan SCD jenis 1:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

-- Create and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Setelah menjalankan contoh SCD tipe 1, tabel target berisi rekaman berikut:

Setelah menjalankan contoh SCD tipe 1 dengan catatan TRUNCATE tambahan, rekaman 124 dan 126 dipotong karena operasi TRUNCATE pada sequenceNum=3, dan tabel target berisi rekaman berikut:

Proses pembaruan SCD tipe 2

Contoh berikut menunjukkan pemrosesan pembaruan SCD tipe 2:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

-- Create and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Setelah menjalankan contoh SCD tipe 2, tabel target berisi rekaman berikut:

Kueri SCD tipe 2 juga dapat menentukan subset kolom output yang akan dilacak untuk riwayat dalam tabel target. Perubahan pada kolom lain diperbarui langsung tanpa menghasilkan catatan riwayat baru. Contoh berikut menunjukkan tidak termasuk kolom city dari pelacakan:

Contoh berikut menunjukkan penggunaan rekam jejak dengan SCD tipe 2:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

-- Create and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Setelah menjalankan contoh ini tanpa catatan TRUNCATE tambahan, tabel target berisi rekaman berikut:

Hasilkan data pengujian

Kode di bawah ini disediakan untuk menghasilkan contoh himpunan data untuk digunakan dalam contoh kueri yang ada dalam tutorial ini. Dengan asumsi Anda memiliki kredensial yang tepat untuk membuat skema baru dan membuat tabel baru, Anda dapat menjalankan pernyataan ini dengan notebook atau Databricks SQL. Kode berikut tidak dimaksudkan untuk dijalankan sebagai bagian dari alur DLT:

CREATE SCHEMA IF NOT EXISTS cdc_data;

CREATE TABLE
  cdc_data.users
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The above batch at sequenceNum 6 will be the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Contoh : Pemrosesan rekam jepret berkala

Contoh berikut menunjukkan pemrosesan SCD tipe 2 yang menyerap rekam jepret tabel yang disimpan di mycatalog.myschema.mytable. Hasil pemrosesan ditulis ke tabel bernama target.

mycatalog.myschema.mytable rekaman pada tanda waktu 2024-01-01 00:00:00

mycatalog.myschema.mytable rekaman pada tanda waktu 2024-01-01 12:00:00

import dlt

@dlt.view(name="source")
def source():
 return spark.read.table("mycatalog.myschema.mytable")

dlt.create_streaming_table("target")

dlt.apply_changes_from_snapshot(
 target="target",
 source="source",
 keys=["key"],
 stored_as_scd_type=2
)

Setelah memproses rekam jepret, tabel target berisi rekaman berikut:

Contoh : Pemrosesan rekam jepret historis

Contoh berikut menunjukkan pemrosesan SCD tipe 2 yang memperbarui tabel target berdasarkan peristiwa sumber dari dua rekam jepret yang disimpan dalam sistem penyimpanan cloud:

Snapshot di timestamp, disimpan di /<PATH>/filename1.csv

Snapshot di timestamp + 5, disimpan di /<PATH>/filename2.csv

Contoh kode berikut menunjukkan pemrosesan pembaruan SCD tipe 2 dengan rekam jepret ini:

import dlt

def exist(file_name):
  # Storage system-dependent function that returns true if file_name exists, false otherwise

# This function returns a tuple, where the first value is a DataFrame containing the snapshot
# records to process, and the second value is the snapshot version representing the logical
# order of the snapshot.
# Returns None if no snapshot exists.
def next_snapshot_and_version(latest_snapshot_version):
  latest_snapshot_version = latest_snapshot_version or 0
  next_version = latest_snapshot_version + 1
  file_name = "dir_path/filename_" + next_version + ".csv"
  if (exist(file_name)):
    return (spark.read.load(file_name), next_version)
   else:
     # No snapshot available
     return None

dlt.create_streaming_live_table("target")

dlt.apply_changes_from_snapshot(
  target = "target",
  source = next_snapshot_and_version,
  keys = ["Key"],
  stored_as_scd_type = 2,
  track_history_column_list = ["TrackingCol"]
)

Setelah memproses rekam jepret, tabel target berisi rekaman berikut:

Menambahkan, mengubah, atau menghapus data dalam tabel streaming target

Jika alur Anda menerbitkan tabel ke Unity Catalog, Anda dapat menggunakan pernyataan bahasa manipulasi data (DML), termasuk pernyataan sisipkan, perbarui, hapus, dan gabungkan, untuk mengubah tabel streaming target yang dibuat oleh pernyataan APPLY CHANGES INTO.

Karena DLT menggunakan kolom SEQUENCE BY tertentu dan menyebarluaskan nilai pengurutan yang sesuai ke kolom __START_AT dan __END_AT tabel target (untuk SCD tipe 2), Anda harus memastikan bahwa pernyataan DML menggunakan nilai yang valid untuk kolom ini untuk mempertahankan urutan rekaman yang tepat. Lihat Bagaimana CDC diterapkan dengan API APPLY CHANGES?.

Untuk informasi selengkapnya tentang menggunakan pernyataan DML dengan tabel streaming, lihat Menambahkan, mengubah, atau menghapus data dalam tabel streaming.

Contoh berikut menyisipkan rekaman aktif dengan urutan awal 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Membaca umpan data perubahan dari tabel target APPLY CHANGES

Di Databricks Runtime 15.2 ke atas, Anda dapat membaca umpan data perubahan dari tabel streaming yang merupakan target kueri APPLY CHANGES atau APPLY CHANGES FROM SNAPSHOT dengan cara yang sama seperti Anda membaca umpan data perubahan dari tabel Delta lainnya. Langkah-langkah berikut diperlukan untuk membaca umpan data perubahan dari tabel streaming target:

• Tabel streaming target harus dipublikasikan ke Katalog Unity. Lihat Gunakan Katalog Unity dengan alur DLT Anda.
• Untuk membaca umpan data perubahan dari tabel streaming target, Anda harus menggunakan Databricks Runtime 15.2 atau lebih tinggi. Untuk membaca umpan data perubahan di alur DLT lain, alur harus dikonfigurasi untuk menggunakan Databricks Runtime 15.2 atau lebih tinggi.

Anda membaca umpan data perubahan dari tabel streaming target yang dibuat dalam alur DLT dengan cara yang sama seperti membaca umpan data perubahan dari tabel Delta lainnya. Untuk mempelajari selengkapnya tentang menggunakan fungsionalitas umpan data perubahan Delta, termasuk contoh di Python dan SQL, lihat Menggunakan umpan data perubahan Delta Lake di Azure Databricks.

Mendapatkan data tentang rekaman yang diproses oleh kueri DLT CDC

Metrik berikut diambil oleh kueri APPLY CHANGES:

• num_upserted_rows: Jumlah baris output yang di-upsert ke dalam himpunan data selama pembaruan.
• num_deleted_rows: Jumlah baris output yang ada yang dihapus dari himpunan data selama pembaruan.

Metrik jenis num_output_rows, keluaran untuk alur non-CDC, tidak direkam pada kueri apply changes.

Objek data apa yang digunakan untuk pemrosesan DLT CDC?

Saat Anda mendeklarasikan tabel target di metastore Apache Hive, dua struktur data dibuat:

• Tampilan menggunakan nama yang ditetapkan ke tabel target.
• Tabel dukungan internal yang digunakan oleh DLT untuk mengelola pemrosesan CDC. Tabel ini dinamai dengan menambahkan __apply_changes_storage_ sebelum nama tabel target.

Misalnya, jika Anda mendeklarasikan tabel target bernama dlt_cdc_target, Anda akan melihat tampilan bernama dlt_cdc_target dan tabel bernama __apply_changes_storage_dlt_cdc_target di metastore. Membuat tampilan memungkinkan DLT untuk memfilter informasi tambahan (misalnya, batu nisan dan versi) yang diperlukan untuk menangani data yang tidak berurutan. Untuk menampilkan data yang diproses, kueri tampilan target. Karena skema tabel __apply_changes_storage_ mungkin berubah untuk mendukung fitur atau penyempurnaan di masa mendatang, Anda tidak boleh mengkueri tabel untuk penggunaan produksi. Jika Anda menambahkan data secara manual ke tabel, rekaman diasumsikan datang sebelum perubahan lain karena kolom versi hilang.