Realtime Compute for Apache Flink:Buat dan gunakan tabel materialized

Batasan

• Anda hanya dapat memodifikasi kueri untuk tabel materialized yang dibuat di VVR 11.1 atau versi yang lebih baru.

• Saat memodifikasi kueri, Anda hanya dapat menambahkan kolom dan memodifikasi logika komputasi. Anda tidak dapat mengubah urutan kolom yang sudah ada atau memodifikasi definisinya.

  Operasi	Didukung	Deskripsi
  Tambahkan kolom baru	Ya	Anda dapat menambahkan kolom baru ke skema sambil mempertahankan urutan kolom yang sudah ada.
  Modifikasi logika komputasi untuk kolom yang sudah ada (tanpa mengubah nama atau tipe datanya)	Ya	Anda dapat memodifikasi logika komputasi, tetapi nama kolom dan tipe data harus tetap sama.
  Ubah urutan kolom yang sudah ada	Tidak	Urutan kolom bersifat tetap. Untuk mengubahnya, Anda harus menghapus dan membuat ulang tabel materialized.
  Modifikasi nama atau tipe data kolom yang sudah ada	Tidak	Anda harus menghapus dan membuat ulang tabel materialized.

Contoh

1. Klik Edit Table dan modifikasi kueri. Kode berikut memberikan contohnya:

   ALTER MATERIALIZED TABLE `paimon`.`default`.`mt-orders`
       AS
       SELECT
         *,
         price * quantity AS total_price
       FROM orders
       WHERE price * quantity > 1000
   ;

2. Klik Preview untuk melihat perbandingan sebelum dan sesudah.

   Kotak dialog Materialized Table Change Details menampilkan perubahan berikut: di area SQL Statement, pernyataan SQL berubah dari SELECT * yang asli menjadi SELECT *, <code data-tag="code" id="code_ee0dc0cd25b8">orders.price * orders.quantity AS total_price, dan menyertakan kondisi WHERE baru <code data-tag="code" id="code_18ba1c5cea54">orders.price * orders.quantity > 1000. Di area Table Columns, bidang total_price (DOUBLE) yang baru disorot dengan warna hijau.

3. Klik OK. Anda dapat melihat kolom yang baru ditambahkan dan logika kueri yang dimodifikasi pada tab Table Schema.

Pembaruan inkremental

CREATE MATERIALIZED TABLE mt_shipped_orders (
    PRIMARY KEY (order_id) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT 
    order_id,
    COALESCE(customer_id, 'Unknown') AS customer_id,
    CAST(order_amount AS DECIMAL(10, 2)) AS order_amount,
    CASE 
        WHEN status = 'shipped' THEN 'Completed'
        WHEN status = 'pending' THEN 'In Progress'
        ELSE 'Unknown'
    END AS order_status,
    DATE_FORMAT(order_ts, 'yyyyMMdd') AS order_date,
    UDSF_ProcessFunction(notes) AS notes
FROM 
    orders
WHERE
    status = 'shipped';

CREATE MATERIALIZED TABLE mt_enriched_orders (
    PRIMARY KEY (order_id, order_tag) NOT ENFORCED
)
FRESHNESS = INTERVAL '30' MINUTE
AS
WITH o AS (
    SELECT
        order_id,
        product_id,
        quantity,
        proc_time,
        e.tag AS order_tag
    FROM 
        orders,
        LATERAL TABLE(UDTF_StringSplitFunction(tags, ',')) AS e(tag))
SELECT 
    o.order_id,
    o.product_id,
    p.product_name,
    p.category,
    o.quantity,
    p.price,
    o.quantity * p.price AS total_amount,
    order_tag
FROM o 
LEFT JOIN 
    product_info FOR SYSTEM_TIME AS OF PROCTIME() AS p
ON 
    o.product_id = p.product_id;

Pengisian ulang data

Sebelumnya, memperbaiki hasil pemrosesan aliran untuk data historis memerlukan pengembangan pekerjaan batch terpisah. Dengan menggunakan tabel materialized, Anda dapat langsung mengisi ulang partisi data historis. Pendekatan ini menyatukan pemrosesan batch dan streaming, sehingga mengurangi biaya pengembangan dan O&M.

1. Klik tabel materialized target di bawah katalognya.

2. Pada tab Data Information, isi ulang data.

   Jika Anda menentukan kolom partisi saat membuat tabel materialized, tabel tersebut adalah tabel partisi. Jika tidak, tabel tersebut adalah tabel non-partisi.

   Tabel partisi

   Di bagian Partitions, klik Trigger Update jika ini pertama kalinya Anda mengisi ulang data atau jika partisi yang diperlukan belum ada. Jika partisi sudah ada, Anda dapat memilih partisi tertentu dan klik Refresh di kolom Actions.

   Klik tab Data Information dan lakukan operasi di area Data Partitions di bagian bawah halaman.

   Parameter

   • Kolom partisi: Kolom partisi tabel. Misalnya, jika Anda memasukkan 20241201, semua data dalam partisi ds=20241201 akan diisi ulang.

   • Nama Tugas: Nama tugas pengisian ulang data.

   • Rentang Pembaruan (Opsional): Menentukan apakah akan melakukan pembaruan kaskade ke tabel materialized hilir. Dimulai dari tabel saat ini, semua tabel materialized dalam alur data akan diperbarui. Kedalaman maksimum alur data hilir yang didukung adalah 6 level.

     Catatan

     • Saat memperbarui tabel partisi, tabel materialized hilir harus memiliki kolom partisi yang persis sama dengan tabel awal. Jika tidak, operasi pembaruan akan gagal.

     • Jika pembaruan gagal untuk tabel materialized apa pun dalam alur data, semua node hilir berikutnya juga akan gagal.

   • Target Penyebaran: Anda dapat memilih antrian atau kluster session. Pilihan default adalah default-queue.

   Tabel non-partisi

   Di bagian Data Detail, klik Refresh.

   Parameter

   • Nama Tugas: Nama tugas pengisian ulang data.

   • Rentang Pembaruan: Opsi ini tidak tersedia untuk tabel non-partisi.

     Catatan

     • Selama pembaruan, data hilir direfresh sepenuhnya.

     • Jika pembaruan gagal untuk tabel materialized apa pun dalam alur data, semua node hilir berikutnya juga akan gagal.

     • Pembaruan kaskade tidak didukung jika tabel awal adalah tabel non-partisi yang diperbarui oleh pekerjaan streaming.

   • Target Penyebaran: Anda dapat memilih antrian atau kluster session. Pilihan default adalah default-queue.

3. Pengisian ulang terjadwal dan batch.

   Anda dapat menggunakan Task Orchestration untuk membuat alur kerja bagi tabel materialized agar menjalankan pekerjaan pengisian ulang secara terjadwal. Anda juga dapat menggunakan fitur pengisian ulang data alur kerja untuk mengisi ulang data secara massal dalam rentang waktu tertentu.

Ubah kesegaran data

1. Di katalog yang sesuai, klik database materialized table, lalu pilih materialized table target.

2. Di pojok kanan atas, klik Modify Freshness.

   • Jika tabel materialized tidak memiliki primary key, Anda tidak dapat mengganti metode pembaruannya antara streaming dan batch. Sistem menggunakan pekerjaan streaming untuk nilai kesegaran data di bawah 30 menit dan pekerjaan batch untuk nilai 30 menit atau lebih. Oleh karena itu, mengubah kesegaran data melewati ambang batas 30 menit ini tidak diizinkan untuk tabel tanpa primary key.

   • Jika tabel hulu adalah tabel materialized, pastikan kesegaran data tabel hilir merupakan kelipatan bilangan bulat positif dari kesegaran data tabel hulu.

   • Kesegaran data maksimum adalah satu hari.

Lihat alur data

Di panel navigasi sebelah kiri, pilih O&M > Data Lineage untuk membuka halaman alur data untuk tabel materialized. Di halaman ini, Anda dapat melihat hubungan alur data antara semua tabel materialized. Anda juga dapat melakukan operasi seperti Start/Stop Update dan Modify Freshness langsung pada tabel materialized. Klik Details untuk membuka halaman detail tabel materialized yang sesuai.

Klik node tabel materialized untuk memperluas panel detailnya. Panel tersebut menampilkan nilai Data Freshness, Latest Update Time, dan Materialized Table Status, serta menyediakan aksi Trigger Update.