Menggunakan tabel streaming mandiri

Tabel streaming mandiri adalah tabel yang terdaftar di Unity Catalog dengan dukungan tambahan untuk pemrosesan data streaming atau inkremental, yang didefinisikan di luar alur Lakeflow. Alur secara otomatis dibuat untuk setiap tabel streaming. Anda dapat menggunakan tabel streaming untuk pemuatan data bertahap dari Kafka dan penyimpanan objek cloud.

Anda dapat membuat dan memuat ulang tabel streaming mandiri dari gudang data Databricks SQL, atau dari notebook yang berjalan pada komputasi umum nirserver. Untuk detail tentang perbedaan antara dua opsi komputasi, lihat Persyaratan untuk alur mandiri.

Untuk membuat dan memuat ulang tabel streaming mandiri dengan Python dari notebook, lihat Gunakan Python dengan pipeline mandiri.

Note

Untuk mempelajari cara menggunakan tabel Delta Lake sebagai sumber streaming dan sink, lihat Baca dan tulis streaming tabel Delta Lake.

Requirements

Untuk opsi komputasi, izin, dan persyaratan lainnya untuk membuat, menyegarkan, dan mengkueri tabel streaming mandiri, lihat Persyaratan untuk alur mandiri.

Membuat tabel streaming

Tabel streaming ditentukan oleh kueri SQL di Databricks SQL. Saat Anda membuat tabel streaming, data yang saat ini ada di tabel sumber digunakan untuk membangun tabel streaming. Setelah itu, Anda me-refresh tabel, biasanya sesuai jadwal, untuk menarik data apa pun yang ditambahkan dalam tabel sumber untuk ditambahkan ke tabel streaming.

Saat membuat tabel streaming, Anda dianggap sebagai pemilik tabel.

Untuk membuat tabel streaming dari tabel yang sudah ada, gunakan CREATE STREAMING TABLE pernyataan , seperti dalam contoh berikut:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT product, price FROM STREAM raw_data;

Dalam hal ini, tabel streaming sales dibuat dari kolom tertentu pada tabel raw_data, dengan jadwal untuk diperbarui setiap jam. Kueri yang digunakan harus berupa kueri streaming . Gunakan kata kunci STREAM untuk memanfaatkan semantik streaming dalam membaca dari sumber.

Komputasi yang digunakan untuk refresh

Saat Anda membuat tabel streaming menggunakan CREATE OR REFRESH STREAMING TABLE pernyataan , refresh data awal dan populasi segera dimulai. Operasi ini tidak menggunakan komputasi gudang Databricks SQL. Sebagai gantinya, tabel streaming mengandalkan alur tanpa server untuk pembuatan dan refresh. Alur tanpa server khusus secara otomatis dibuat dan dikelola oleh sistem untuk setiap tabel streaming.

Memuat file dengan Auto Loader

Untuk membuat tabel streaming dari file dalam volume, Anda menggunakan Auto Loader. Gunakan Auto Loader untuk sebagian besar tugas penyerapan data dari penyimpanan objek cloud. Auto Loader dan pipeline dirancang untuk memuat data yang terus bertambah secara inkremental dan idempoten saat tiba di penyimpanan cloud.

Untuk menggunakan Auto Loader di Databricks SQL, gunakan fungsi .read_files Contoh berikut menunjukkan penggunaan Auto Loader untuk membaca volume file JSON ke dalam tabel streaming:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT * FROM STREAM read_files(
    "/Volumes/my_catalog/my_schema/my_volume/path/to/data",
    format => "json"
  );

Untuk membaca data dari penyimpanan cloud, Anda juga dapat menggunakan Auto Loader:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
    format => "json"
  );

Untuk mempelajari tentang Auto Loader, lihat Apa itu Auto Loader?. Untuk mempelajari selengkapnya tentang menggunakan Auto Loader di SQL, dengan contoh, lihat Memuat data dari penyimpanan objek.

Penyerapan streaming dari sumber lain

Misalnya pengambilan data dari sumber lain, termasuk Kafka, dapat dilihat pada Memuat data melalui pipa pemrosesan.

Menerapkan penangkapan data perubahan (CDC) dengan alur CDC Otomatis

Gunakan klausul FLOW AUTO CDC untuk memproses rekaman *change data capture* (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. Lihat Mengubah pengambilan data dan rekam jepret.

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).

Contoh berikut membuat tabel streaming yang menerapkan perubahan CDC menggunakan SCD tipe 1:

CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

Contoh berikut menggunakan SCD tipe 2 untuk mempertahankan riwayat perubahan:

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;

Untuk detail selengkapnya tentang opsi dan perilaku CDC Otomatis, lihat API CDC OTOMATIS: Menyederhanakan perubahan pengambilan data dengan alur. Untuk referensi sintaks lengkap, lihat CREATE STREAMING TABLE.

Terapkan penggantian batch selektif dengan alur kerja REPLACE WHERE

Gunakan klausul FLOW REPLACE WHERE untuk menghitung ulang dan menimpa subset tertentu dari tabel streaming tanpa memproses ulang seluruh riwayat tabel. REPLACE WHERE alur kerja sangat cocok untuk pemrosesan batch inkremental untuk join dan agregasi, data yang datang terlambat, pemrosesan ulang di hulu, evolusi skema, dan pengisian data balik.

Untuk detail selengkapnya tentang alur REPLACE WHERE, termasuk persyaratan, penimpaan predikat, dan penyegaran inkremental, lihat alur REPLACE WHERE untuk tabel streaming mandiri.

Terapkan penggantian snapshot parsial dengan alur REPLACE USING

Important

Alur kerja GANTI MENGGUNAKAN masih dalam Beta.

Gunakan klausa ini FLOW REPLACE USING untuk menjaga tabel streaming tetap sinkron dengan aliran snapshot parsial. Pada setiap pembaruan, alur REPLACE USING menggantikan semua baris yang sesuai dengan kolom kunci yang ditentukan dan membiarkan semua baris lainnya tetap sama. Sebuah kolom mengatur SEQUENCE BY pembaruan sehingga urutan tertinggi untuk sebuah kunci selalu menang, bahkan ketika pembaruan datang tidak sesuai urutan. Contohnya:

CREATE OR REFRESH STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

BY NAME diperlukan. Ini mencocokkan kolom berdasarkan nama, bukan berdasarkan posisi.

REPLACE USING berperilaku sama untuk tabel streaming mandiri seperti pada pipeline Lakeflow. Untuk mengetahui cara kerja, urutan, hal-hal yang diharapkan, keterbatasan, dan contohnya, lihat Penggantian snapshot sebagian dengan alur REPLACE USING. Perbedaan berikut berlaku untuk tabel streaming mandiri:

  • Definisikan alur dalam SQL. Buat alur REPLACE USING dengan klausa SQL inline FLOW REPLACE USING pada CREATE OR REFRESH STREAMING TABLE. Pernyataan terpisah CREATE FLOW adalah konstruk pipeline Lakeflow dan tidak digunakan untuk tabel streaming terpisah.
  • Komputasi dikelola untuk Anda. Tabel streaming mandiri berjalan di pipeline serverless yang dikelola oleh sistem dan memerlukan Databricks Runtime 18.2 atau yang lebih baru. Anda tidak bisa memilih antara komputasi klasik dan serverless.

Menyerap data baru saja

Secara default, read_files fungsi membaca semua data yang ada di folder sumber selama pembuatan tabel, lalu memproses rekaman yang baru tiba dengan setiap refresh.

Untuk menghindari penyerapan data yang sudah ada di folder sumber pada saat pembuatan tabel, atur includeExistingFiles opsi ke false. Ini berarti bahwa hanya data yang tiba di folder setelah pembuatan tabel diproses. Contohnya:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    '/path/to/files',
    includeExistingFiles => false
  );

Versi runtime

Tabel streaming selalu berjalan pada versi runtime SQL Databricks terbaru. Properti tabel pipelines.channel, yang sebelumnya digunakan untuk memilih kanal runtime preview atau current, tidak lagi didukung dan tidak berpengaruh. Jika definisi yang sudah ada mencakup properti ini, maka definisi tersebut diabaikan dengan aman dan Anda tidak perlu menghapusnya.

Menyembunyikan data sensitif

Anda dapat menggunakan tabel streaming untuk menyembunyikan data sensitif dari pengguna yang mengakses tabel. Salah satu pendekatannya adalah menentukan kueri sehingga mengecualikan kolom atau baris sensitif sepenuhnya. Atau, Anda bisa menerapkan masker kolom atau filter baris berdasarkan izin pengguna kueri. Misalnya, Anda mungkin menyembunyikan tax_id kolom untuk pengguna yang tidak berada dalam grup HumanResourcesDept. Untuk melakukan ini, gunakan sintaks ROW FILTER dan MASK selama pembuatan tabel streaming. Untuk informasi selengkapnya, lihat Filter baris dan masker kolom.

Memperbarui tabel streaming

Tabel streaming secara otomatis membuat dan menggunakan alur tanpa server untuk memproses operasi refresh. Refresh dikelola oleh pipeline dan pembaruan dipantau oleh gudang SQL Databricks yang digunakan untuk membuat tabel streaming. Tabel streaming dapat diperbarui menggunakan alur yang berjalan sesuai jadwal.

Bahkan jika Anda memiliki refresh terjadwal, Anda dapat memanggil refresh manual kapan saja. Refresh ditangani oleh alur yang sama yang secara otomatis dibuat bersama dengan tabel streaming.

Untuk me-refresh tabel streaming:

REFRESH STREAMING TABLE sales;

Anda dapat memeriksa status refresh terbaru dengan DESCRIBE TABLE EXTENDED.

Note

Anda mungkin perlu me-refresh tabel streaming sebelum menggunakan kueri perjalanan waktu .

Untuk mempelajari cara menjadwalkan refresh, lihat Menjadwalkan refresh. Penyegaran terjadwal dapat memiliki pemberitahuan pembaruan, dan Anda dapat mengatur mode kinerja untuk penyegaran.

Cara kerja refresh

Refresh tabel streaming hanya memeriksa baris baru yang telah tiba setelah pembaruan terakhir, dan hanya menambahkan data baru.

Setiap refresh menggunakan definisi tabel streaming saat ini untuk memproses data baru ini. Memodifikasi definisi tabel streaming tidak secara otomatis menghitung ulang data yang ada. Jika modifikasi tidak kompatibel dengan data yang ada (misalnya, mengubah jenis data), refresh berikutnya gagal dengan kesalahan.

Contoh berikut menjelaskan bagaimana perubahan pada definisi tabel streaming memengaruhi perilaku refresh:

  • Menghapus filter tidak memproses ulang baris yang difilter sebelumnya.
  • Mengubah proyeksi kolom tidak memengaruhi bagaimana data yang ada diproses.
  • Penggabungan dengan rekam jepret statis menggunakan status rekam jepret pada waktu pemrosesan pertama kali. Data yang terlambat tiba yang akan cocok dengan rekam jepret yang diperbarui diabaikan. Hal ini dapat menyebabkan fakta dihilangkan jika dimensi terlambat.
  • Modifikasi pada fungsi CAST dari kolom yang sudah ada dapat menyebabkan kesalahan.

Jika data Anda berubah dengan cara yang tidak dapat didukung dalam tabel streaming yang ada, Anda dapat melakukan refresh penuh.

Memperbarui sepenuhnya tabel streaming

Refresh penuh memproses ulang semua data yang tersedia di sumber dengan definisi terbaru. Tidak disarankan untuk memanggil refresh penuh pada sumber yang tidak menyimpan seluruh riwayat data atau memiliki periode retensi singkat, seperti Kafka, karena refresh penuh memotong data yang ada. Anda mungkin tidak dapat memulihkan data lama jika data tidak lagi tersedia di sumbernya.

Contohnya:

REFRESH STREAMING TABLE sales FULL;

Menjadwalkan dan memantau refresh

Anda dapat menyegarkan tabel streaming secara otomatis sesuai jadwal atau saat data upstream berubah, dan Anda dapat mengonfigurasi batas waktu refresh, pemberitahuan, dan mode performa. Lihat Jadwal penyegaran.

Mengontrol akses ke tabel streaming

Tabel streaming mendukung kontrol akses yang kaya untuk mendukung berbagi data sambil menghindari mengekspos data yang berpotensi privat. Pemilik tabel streaming atau pengguna dengan MANAGE hak istimewa dapat memberikan SELECT hak istimewa kepada pengguna lain. Pengguna dengan akses SELECT ke tabel streaming tidak memerlukan akses SELECT ke tabel yang dirujuk oleh tabel streaming. Kontrol akses ini memungkinkan berbagi data sambil mengontrol akses ke data yang mendasar.

Anda juga dapat memodifikasi pemilik tabel streaming.

Memberikan otoritas pada tabel streaming

Untuk memberikan akses ke tabel streaming, gunakan GRANT pernyataan :

GRANT <privilege_type> ON <st_name> TO <principal>;

Dapat privilege_type berupa:

  • SELECT - pengguna dapat SELECT tabel streaming.
  • REFRESH - pengguna dapat REFRESH tabel streaming. Refresh dijalankan menggunakan izin pemilik.

Contoh berikut membuat tabel streaming dan memberikan hak istimewa pilih dan refresh kepada pengguna:

CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM STREAM source_table;

-- Grant read-only access:
GRANT SELECT ON st_name TO read_only_user;

-- Grant read and refresh access:
GRANT SELECT ON st_name TO refresh_user;
GRANT REFRESH ON st_name TO refresh_user;

Untuk informasi selengkapnya tentang memberikan hak istimewa pada objek yang dapat diamankan Katalog Unity, lihat Referensi hak istimewa Katalog Unity.

Mencabut izin dari tabel streaming

Untuk mencabut akses dari tabel streaming, gunakan REVOKE pernyataan :

REVOKE privilege_type ON <st_name> FROM principal;

Ketika SELECT hak istimewa pada tabel sumber dicabut dari pemilik tabel streaming atau pengguna lain yang telah diberikan MANAGE atau SELECT hak istimewa pada tabel streaming, atau tabel sumber dihilangkan, pemilik tabel streaming atau akses yang diberikan pengguna masih dapat mengkueri tabel streaming. Namun, perilaku berikut terjadi:

  • Pemilik tabel streaming atau orang lain yang kehilangan akses ke tabel streaming tidak dapat lagi REFRESH tabel streaming tersebut, dan tabel streaming menjadi basi dari waktu ke waktu.
  • Jika diotomatisasi dengan jadwal, jadwal berikutnya REFRESH gagal atau tidak dijalankan.

Contoh berikut mencabut SELECT hak istimewa dari read_only_user:

REVOKE SELECT ON st_name FROM read_only_user;

Mengubah pemilik tabel streaming

Pengguna yang memiliki izin MANAGE pada tabel streaming independen dapat menetapkan pemilik baru melalui Catalog Explorer. Pemilik baru dapat menjadi diri mereka sendiri atau perwakilan layanan tempat mereka memiliki peran Pengguna Perwakilan Layanan .

  1. Dari ruang kerja Azure Databricks Anda, klik Ikon data.Katalog untuk membuka Catalog Explorer.

  2. Pilih tabel streaming yang ingin Anda perbarui.

  3. Di bilah sisi kanan, di bawah Tentang tabel streaming ini, temukan Pemilik, dan klik ikon Pensil. edit.

    Note

    Jika Anda mendapatkan pesan yang meminta Anda memperbarui pemilik dengan mengubah pengguna Run as di pengaturan pipeline, berarti tabel streaming didefinisikan dalam pipeline Lakeflow, bukan sebagai tabel mandiri. Pesan menyertakan tautan ke pengaturan alur, tempat Anda dapat mengubah Jalankan sebagai pengguna.

  4. Pilih pemilik baru untuk tabel streaming.

    Pemilik secara otomatis memiliki hak istimewa MANAGE dan SELECT pada tabel streaming yang mereka miliki. Jika Anda mengatur perwakilan layanan sebagai pemilik untuk tabel streaming yang Anda miliki, dan Anda tidak secara eksplisit memiliki SELECT atau MANAGE hak istimewa pada tabel streaming, maka perubahan ini akan menyebabkan Anda kehilangan semua akses ke tabel streaming. Dalam hal ini, Anda diminta untuk secara eksplisit memberikan hak istimewa tersebut.

    Pilih Grant MANAGE dan Berikan SELECT hak istimewa agar hak istimewa tersebut tersedia di Simpan.

  5. Klik Simpan untuk mengubah pemilik.

Tabel streaming kini memiliki pemilik baru. Semua pembaruan di masa mendatang dijalankan menggunakan identitas pemilik baru.

Ketika pemilik kehilangan hak istimewa untuk tabel sumber

Jika Anda mengubah pemilik, dan pemilik baru tidak mempunyai akses ke tabel sumber (atau izin akses dicabut pada tabel sumber yang mendasar), pengguna masih dapat melakukan kueri pada tabel streaming. Namun:

  • Mereka tidak dapat REFRESH melakukan streaming tabel.
  • Refresh terjadwal berikutnya dari tabel streaming gagal.

Kehilangan akses ke data sumber mencegah adanya pembaruan, tetapi tidak segera menghentikan pembacaan tabel streaming yang ada.

Menghapus rekaman secara permanen dari tabel streaming

Important

Dukungan untuk pernyataan REORG dengan tabel streaming ada di Pratinjau Umum .

Note

  • Menggunakan pernyataan REORG dengan tabel streaming memerlukan Databricks Runtime 15.4 ke atas.
  • Meskipun Anda dapat menggunakan REORG pernyataan dengan tabel streaming apa pun, itu hanya diperlukan saat menghapus rekaman dari tabel streaming dengan vektor penghapusan diaktifkan. Perintah tidak berpengaruh saat digunakan dengan tabel streaming tanpa vektor penghapusan diaktifkan.

Untuk menghapus catatan secara fisik dari penyimpanan yang digunakan untuk tabel streaming dengan fitur vektor penghapusan yang diaktifkan, seperti untuk kepatuhan GDPR, diperlukan langkah tambahan untuk memastikan bahwa operasi VACUUM dilakukan pada data tabel streaming.

Untuk menghapus rekaman secara fisik dari penyimpanan yang mendasar:

  1. Memperbarui rekaman atau menghapus rekaman dari tabel streaming.
  2. Jalankan pernyataan REORG terhadap tabel streaming, menentukan parameter APPLY (PURGE). Contoh: REORG TABLE <streaming-table-name> APPLY (PURGE);.
  3. Tunggu hingga periode retensi data tabel streaming berlalu. Periode retensi data default adalah tujuh hari, tetapi dapat dikonfigurasi dengan properti tabel delta.deletedFileRetentionDuration. Lihat Mengonfigurasi retensi data untuk kueri perjalanan lintas waktu.
  4. REFRESH tabel streaming. Lihat Penyegaran tabel streaming. Dalam waktu 24 jam setelah operasi REFRESH, tugas pemeliharaan pipa saluran, termasuk operasi VACUUM yang dibutuhkan untuk memastikan catatan dihapus secara permanen, dieksekusi secara otomatis.

Memantau eksekusi menggunakan riwayat kueri

Anda bisa menggunakan halaman riwayat kueri untuk mengakses detail kueri dan profil kueri yang dapat membantu Anda mengidentifikasi kueri yang kinerjanya buruk dan kendala dalam alur yang digunakan untuk melakukan pembaruan tabel streaming Anda. Untuk gambaran umum jenis informasi yang tersedia dalam riwayat kueri dan profil kueri, lihat Riwayat kueri dan Profil kueri.

Important

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

Semua pernyataan yang terkait dengan tabel streaming muncul dalam riwayat kueri. Anda dapat menggunakan filter drop-down Pernyataan untuk memilih perintah apa pun dan memeriksa kueri terkait. Semua CREATE pernyataan diikuti oleh REFRESH pernyataan yang dijalankan secara asinkron di dalam jalur pemrosesan. Pernyataan REFRESH biasanya mencakup rencana kueri terperinci yang memberikan wawasan tentang mengoptimalkan performa.

Untuk mengakses REFRESH pernyataan di antarmuka pengguna riwayat kueri, gunakan langkah-langkah berikut:

  1. Klik ikon Riwayat. di bilah sisi kiri untuk membuka UI Riwayat Kueri .
  2. Pilih kotak centang REFRESH dari filter dropdown Pernyataan.
  3. Klik nama pernyataan kueri untuk menampilkan detail ringkasan seperti durasi kueri dan metrik agregat.
  4. Klik Lihat profil kueri untuk membuka profil kueri. Lihat Profil kueri untuk detail tentang menavigasi profil kueri.
  5. Secara opsional, Anda bisa menggunakan tautan di bagian Sumber Kueri untuk membuka kueri atau alur terkait.

Anda juga dapat mengakses detail kueri menggunakan tautan di editor SQL atau dari buku catatan yang dilampirkan ke gudang SQL.

Mengakses tabel streaming dari klien eksternal

Untuk mengakses tabel streaming dari klien Delta Lake atau Iceberg eksternal yang tidak mendukung API terbuka, Anda dapat menggunakan Mode Kompatibilitas. Mode Kompatibilitas membuat versi baca-saja dari tabel streaming Anda yang dapat diakses oleh klien Delta Lake atau Iceberg apa pun.

Sumber daya tambahan

  • Terakhir diperbarui pada