create_auto_cdc_from_snapshot_flow

Fungsi ini create_auto_cdc_from_snapshot_flow membuat alur yang menggunakan alur Lakeflow mengubah fungsionalitas pengambilan data (CDC) untuk memproses data sumber dari rekam jepret database. Lihat Cara kerja AUTO CDC FROM SNAPSHOT.

Nota

Fungsi ini menggantikan fungsi apply_changes_from_snapshot()sebelumnya . Kedua fungsi memiliki tanda tangan yang sama. Databricks merekomendasikan pembaruan untuk menggunakan nama baru.

Penting

Anda harus memiliki tabel streaming target untuk operasi ini. Untuk membuat tabel target yang diperlukan, Anda bisa menggunakan fungsi create_streaming_table().

Syntax

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Nota

Untuk AUTO CDC FROM SNAPSHOT pemrosesan, perilaku defaultnya adalah menyisipkan baris baru saat rekaman yang cocok dengan kunci yang sama tidak ada di target. Jika rekaman yang cocok memang ada, rekaman tersebut hanya diperbarui jika salah satu nilai dalam baris telah berubah. Baris dengan kunci yang ada di target tetapi tidak lagi ada di sumber akan dihapus.

Untuk mempelajari selengkapnya tentang pemrosesan CDC dengan rekam jepret, lihat API CDC OTOMATIS: Menyederhanakan perubahan pengambilan data dengan alur. Untuk contoh penggunaan fungsi ini create_auto_cdc_from_snapshot_flow() , lihat contoh pengambilan snapshot periodik, pengambilan snapshot historis, dan backfill SCD Type 1 .

Parameter-parameternya

Pengaturan	Tipe	Description
`target`	`str`	Dibutuhkan. Nama tabel yang akan diperbarui. Anda dapat menggunakan fungsi create_streaming_table() untuk membuat tabel target sebelum menjalankan `create_auto_cdc_from_snapshot_flow()` fungsi.
`source`	`str` atau `lambda function`	Dibutuhkan. Baik nama tabel atau view untuk diambil cuplikannya secara berkala atau fungsi lambda Python yang mengembalikan DataFrame cuplikan yang akan diproses dan versi cuplikan tersebut. Lihat Menerapkan `source` argumen.
`keys`	`list`	Dibutuhkan. Kolom atau kombinasi kolom yang secara unik mengidentifikasi baris dalam data sumber. Ini digunakan untuk mengidentifikasi peristiwa CDC mana yang berlaku untuk rekaman tertentu dalam tabel target. Anda dapat menentukan: Daftar string: `["userId", "orderId"]` Daftar fungsi Spark SQL `col()` : `[col("userId"), col("orderId"]`. Argumen ke `col()` fungsi tidak dapat menyertakan kualifikasi. Misalnya, Anda dapat menggunakan `col(userId)`, tetapi Anda tidak dapat menggunakan `col(source.userId)`.
`stored_as_scd_type`	`str` atau `int`	Apakah akan menyimpan rekaman sebagai SCD tipe 1 atau SCD tipe 2. Atur ke `1` untuk SCD tipe 1 atau `2` untuk SCD tipe 2. Defaultnya adalah SCD tipe 1.
`track_history_column_list` atau `track_history_except_column_list`	`list`	Sebagian kolom keluaran yang perlu dilacak untuk riwayat dalam tabel sasaran. Gunakan `track_history_column_list` untuk menentukan daftar lengkap kolom yang akan dilacak. Gunakan `track_history_except_column_list` untuk menentukan kolom yang akan dikecualikan dari pelacakan. Anda dapat mendeklarasikan nilai sebagai daftar string atau sebagai fungsi Spark SQL `col()` : `track_history_column_list = ["userId", "name", "city"]` `track_history_column_list = [col("userId"), col("name"), col("city")]` `track_history_except_column_list = ["operation", "sequenceNum"]` `track_history_except_column_list = [col("operation"), col("sequenceNum")` Argumen ke `col()` fungsi tidak dapat menyertakan kualifikasi. Misalnya, Anda dapat menggunakan `col(userId)`, tetapi Anda tidak dapat menggunakan `col(source.userId)`. Secara default, semua kolom dalam tabel target disertakan ketika tidak ada argumen `track_history_column_list` atau `track_history_except_column_list` diberikan ke fungsi.
`once`	`bool`	Apakah akan menjalankan alur snapshot hanya satu kali. Ketika `True`, alur dilewati selama pembaruan bertahap berikutnya setelah berhasil dikomitmen. Penyegaran penuh target akan menjalankan ulang alur tersebut. Gunakan opsi ini untuk memuat snapshot satu kali atau mengisi ulang. Defaultnya adalah `False`.

Notes

Untuk target SCD Tipe 1 dalam pipeline yang terpicu, satu create_auto_cdc_from_snapshot_flow() alur dapat berbagi target dengan satu atau lebih AUTO CDC alur yang didefinisikan dalam Python atau SQL. Persyaratan berikut berlaku:

Berikan nama unik pada setiap AUTO CDC aliran.
Gunakan jumlah kunci yang sama dalam urutan yang sama untuk semua aliran. Nama kunci aliran snapshot dibandingkan secara tidak sensitif dengan huruf besar dan kecil dengan AUTO CDC nama kunci. Beberapa AUTO CDC alur harus menggunakan nama kunci dan casing yang identik.
Gunakan tipe data yang persis sama untuk versi snapshot dan kolom sequencing dari setiap AUTO CDC aliran.
Jangan tentukan ekspektasi pada AUTO CDC FROM SNAPSHOT aliran.
Jangan gunakan IGNORE NULL UPDATES. Di Python, jangan atur ignore_null_updates, ignore_null_updates_column_list, atau ignore_null_updates_except_column_list pada create_auto_cdc_flow() alur.
Jika Anda menambahkan AUTO CDC alur ke target yang sudah ada AUTO CDC FROM SNAPSHOT , target tersebut tidak boleh berisi kolom pengguna yang namanya bertentangan dengan kolom sistem yang dicadangkan AUTO CDC .
Jangan gunakan pola ini untuk target SCD Tipe 2 atau bitemporal.

Sebagai contoh, lihat Tambahkan pengisian ke tabel AUTO CDC SCD Type 1.

`source` Menerapkan argumen

Fungsi create_auto_cdc_from_snapshot_flow() mengandung argumen source. Untuk memproses rekam jepret historis, source argumen diharapkan menjadi fungsi lambda Python yang mengembalikan dua nilai ke create_auto_cdc_from_snapshot_flow() fungsi: Python DataFrame yang berisi data rekam jepret yang akan diproses dan versi rekam jepret.

Pemanggilan pertama fungsi harus mengembalikan DataFrame dan versi snapshot. Jika panggilan pertama kembali None, pembaruan pipeline gagal. Setelah setidaknya satu snapshot diproses, kembalikan None untuk menunjukkan bahwa tidak ada snapshot tambahan yang tersedia.

Berikut ini adalah tanda tangan fungsi lambda:

lambda Any => Optional[(DataFrame, Any)]

Argumen untuk fungsi lambda adalah versi rekam jepret yang terakhir diproses.
Nilai pengembalian fungsi lambda adalah None atau tuple dari dua nilai: Nilai pertama tuple adalah DataFrame yang berisi rekam jepret yang akan diproses. Nilai kedua dari tuple adalah versi cuplikan yang mewakili urutan logis dari cuplikan tersebut.

Contoh yang mengimplementasikan dan memanggil fungsi lambda:

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Runtime alur Lakeflow melakukan langkah-langkah berikut setiap kali alur yang berisi fungsi dipicu create_auto_cdc_from_snapshot_flow() :

next_snapshot_and_version Menjalankan fungsi untuk memuat DataFrame rekam jepret berikutnya dan versi rekam jepret yang sesuai.
Jika pemanggilan pertama tidak mengembalikan DataFrame, pembaruan gagal. Jika pemanggilan berikutnya tidak mengembalikan DataFrame, run berakhir dan pembaruan pipeline ditandai sebagai selesai.
Mendeteksi perubahan dalam rekam jepret baru dan menerapkannya secara bertahap ke tabel target.
Kembali ke langkah #1 untuk memuat rekam jepret berikutnya dan versinya.

Saran dan Komentar

Apakah halaman ini membantu?

Terakhir diperbarui pada 2026-10-05