Klausul ALIRAN AUTO CDC

Berlaku untuk: Databricks SQL

FLOW AUTO CDC Gunakan klausa dengan CREATE STREAMING TABLE untuk memproses perubahan rekaman pengambilan data (CDC) dari sumber ke dalam tabel streaming.

Sebelumnya, pernyataan ini 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 kompleks untuk mengurutkan ulang rekaman.

AUTO CDC menyederhanakan CDC dengan menangani rekaman yang tidak berurutan secara otomatis. Anda menentukan kunci untuk mengidentifikasi rekaman, kolom urutan untuk pemesanan, dan apakah akan menyimpan hasil sebagai SCD tipe 1 (pembaruan langsung) atau SCD tipe 2 (pelacakan riwayat).

Sintaksis

  FLOW AUTO CDC
  FROM source
  KEYS ( keys )
  [ IGNORE NULL UPDATES [ ON { columnList | * EXCEPT ( exceptColumnList ) } ] ]
  [ APPLY AS DELETE WHEN condition ]
  [ APPLY AS TRUNCATE WHEN condition ]
  SEQUENCE BY orderByColumn
  [ SYSTEM SEQUENCE BY systemOrderByColumn ]
  [ COLUMNS { columnList | * EXCEPT ( exceptColumnList ) } ]
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL } ]
  [ TRACK HISTORY ON { columnList | * EXCEPT ( exceptColumnList ) } ]
  [ COLUMNS TO UPDATE columnName ]

Perilaku default untuk INSERT peristiwa dan UPDATE adalah meningkatkan peristiwa CDC dari sumbernya: memperbarui baris apa pun dalam tabel target yang cocok dengan kunci yang ditentukan atau menyisipkan baris baru saat rekaman yang cocok tidak ada di tabel target. Penanganan untuk DELETE peristiwa dapat ditentukan dengan kondisi APPLY AS DELETE WHEN.

Parameter-parameternya

• source

  Sumber data. Sumber harus berupa sumber streaming. Gunakan kata kunci STREAM untuk memanfaatkan semantik streaming dalam membaca dari sumber. Jika pembacaan mengalami perubahan atau penghapusan pada rekaman yang ada, akan menghasilkan kesalahan. Paling aman untuk membaca dari sumber statis atau yang hanya bisa ditambahkan.

  Untuk informasi selengkapnya tentang data streaming, lihat Mengubah data dengan alur.

• KEYS

  Kolom atau kombinasi kolom yang secara unik mengidentifikasi baris dalam data sumber. Nilai dalam kolom ini digunakan untuk mengidentifikasi peristiwa CDC mana yang berlaku untuk rekaman tertentu dalam tabel target.

  Untuk menentukan kombinasi kolom, gunakan daftar kolom yang dipisahkan koma.

  Klausa ini diperlukan.

• IGNORE NULL UPDATES

  Memungkinkan pengambilan pembaruan yang berisi sebagian dari kolom target. Saat peristiwa CDC cocok dengan baris yang ada dan IGNORE NULL UPDATES ditentukan, kolom dengan null nilai mempertahankan nilai yang ada dalam target. Ini juga berlaku untuk kolom bersarang dengan nilai null.

  Untuk pembaruan parsial, tambahkan ON klausa untuk mengontrol kolom mana yang mengabaikan null nilai:

  • IGNORE NULL UPDATES ON columnList: hanya kolom yang tercantum yang mempertahankan nilai yang ada saat nilai masuk adalah null. Semua kolom lainnya menerapkan nilai eksplisit null .
  • IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList): semua kolom kecuali yang tercantum mempertahankan nilai yang ada saat nilai masuk adalah null. Kolom yang tercantum menerapkan nilai eksplisit null .

  Untuk informasi selengkapnya, lihat Menerapkan pembaruan parsial.

  Klausa ini bersifat opsional.

  Secara default, menimpa kolom yang ada dengan nilai null.

• APPLY AS DELETE WHEN

  Menentukan kapan peristiwa CDC harus diperlakukan sebagai DELETE bukan upsert.

  Untuk sumber SCD tipe 2, untuk menangani data yang tidak berurutan, baris yang dihapus untuk sementara dipertahankan sebagai batu nisan dalam tabel Delta yang mendasar, dan tampilan dibuat di metastore yang memfilter batu nisan ini. Interval retensi dapat dikonfigurasi dengan pipelines.cdc.tombstoneGCThresholdInSecondsproperti tabel.

  Klausa ini bersifat opsional.

• APPLY AS TRUNCATE WHEN

  Menentukan kapan peristiwa CDC harus diperlakukan sebagai tabel lengkap TRUNCATE. Karena klausa ini memicu pemotongan penuh tabel target, klausul ini harus digunakan hanya untuk kasus penggunaan tertentu yang memerlukan fungsionalitas ini.

  Klausa APPLY AS TRUNCATE WHEN hanya didukung untuk SCD tipe 1. SCD tipe 2 tidak mendukung operasi pemotongan.

  Klausa ini bersifat opsional.

• SEQUENCE BY

  Nama kolom yang menentukan urutan logis peristiwa CDC dalam data sumber. Pemrosesan alur menggunakan urutan ini untuk menangani peristiwa perubahan yang tiba tidak berurutan.

  Jika beberapa kolom diperlukan untuk pengurutan, gunakan STRUCT ekspresi: kolom akan diurutkan berdasarkan bidang struct pertama terlebih dahulu, lalu dengan bidang kedua jika ada dasi, dan sebagainya.

  Kolom yang ditentukan harus tipe data yang dapat diurutkan.

  Klausa ini diperlukan.

• SYSTEM SEQUENCE BY

  Important

  Bitemporal AUTO CDC berada di Beta.

  Nama kolom yang menentukan waktu sistem di mana setiap peristiwa CDC diketahui oleh sistem. Digunakan dengan STORED AS BITEMPORAL untuk melacak perubahan di seluruh waktu bisnis (SEQUENCE BY) dan waktu sistem. Lihat CdC OTOMATIS Bitemporal.

  Kolom yang ditentukan harus tipe data yang dapat diurutkan.

  Klausa ini bersifat opsional dan hanya berlaku untuk tabel bitemporal.

• COLUMNS

  Menentukan subset kolom untuk disertakan dalam tabel target. Anda dapat:

  • Tentukan daftar lengkap kolom yang akan disertakan: COLUMNS (userId, name, city).
  • Tentukan daftar kolom yang akan dikecualikan: COLUMNS * EXCEPT (operation, sequenceNum)

  Klausa ini bersifat opsional.

  Defaultnya adalah menyertakan semua kolom dalam tabel target saat COLUMNS klausul tidak ditentukan.

• STORED AS

  Apakah akan menyimpan rekaman sebagai SCD tipe 1, SCD tipe 2, atau bitemporal.

  Atur ke BITEMPORAL untuk melacak perubahan di seluruh waktu bisnis dan waktu sistem. Bitemporal membutuhkan SYSTEM SEQUENCE BY dan berada di Beta. Lihat CdC OTOMATIS Bitemporal.

  Klausa ini bersifat opsional.

  Defaultnya adalah SCD tipe 1.

• TRACK HISTORY ON

  Menentukan subset kolom output untuk menghasilkan rekaman riwayat saat ada perubahan pada kolom yang ditentukan. Anda dapat:

  • Tentukan daftar lengkap kolom yang akan dilacak: COLUMNS (userId, name, city).
  • Tentukan daftar kolom yang akan dikecualikan dari pelacakan: COLUMNS * EXCEPT (operation, sequenceNum)

  Klausa ini bersifat opsional. Defaultnya adalah melacak riwayat untuk semua kolom output ketika ada perubahan apa pun, setara dengan TRACK HISTORY ON *.

• COLUMNS TO UPDATE

  Menentukan nama kolom sumber yang menyimpan, untuk setiap rekaman perubahan, kumpulan kolom yang akan diperbarui sebagai array string nama kolom (array<string>). Kolom yang tidak ada dalam array menyimpan nilai target yang ada, sementara kolom yang tercantum ditulis dari sumbernya, termasuk nilai eksplisit null .

  Gunakan klausa ini untuk pembaruan parsial saat setiap rekaman perubahan memperbarui sekumpulan kolom yang berbeda dan Anda perlu menerapkan nilai eksplisit null .

  Anda tidak dapat menggunakan COLUMNS TO UPDATE bersama dengan IGNORE NULL UPDATES, dan tidak didukung untuk tabel bitemporal.

  Untuk informasi selengkapnya, lihat Menerapkan pembaruan parsial.

  Klausa ini bersifat opsional.

Examples

-- SCD type 1: apply CDC changes with direct updates (no history)
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

-- SCD type 2: retain a history of changes, with delete handling
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  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;

-- SCD type 2 with history tracking on specific columns
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  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);

-- SCD type-2 dimension with a generated surrogate key and automatic clustering
> CREATE OR REFRESH STREAMING TABLE ${target_catalog}.silver.dim_customer (
    customer_sk BIGINT GENERATED ALWAYS AS IDENTITY,
    customer_id INT,
    email       STRING,
    segment     STRING,
    city        STRING,
    __START_AT  TIMESTAMP,
    __END_AT    TIMESTAMP
  )
  COMMENT 'SCD Type 2 customer dimension with generated surrogate key'
  CLUSTER BY AUTO
  FLOW AUTO CDC
  FROM stream(${target_catalog}.bronze.customer_changes)
  KEYS (customer_id)
  SEQUENCE BY updated_at
  STORED AS SCD TYPE 2;

Artikel terkait

• CREATE STREAMING TABLE
• API CDC OTOMATIS: Menyederhanakan perubahan pengambilan data dengan alur
• AUTO CDC INTO (pipeline)
• CREATE FLOW (jalur pemrosesan)
• Mengalir