Realtime Compute for Apache Flink:Akses Serverless StarRocks ke Fluss (Beta)

Langkah 1: Buat katalog Fluss

1. Login ke konsol EMR Serverless StarRocks. Pada halaman daftar instans, klik Connect Instance di kolom Actions pada instans target. Untuk detailnya, lihat Connect to a StarRocks instance through EMR StarRocks Manager.

2. Pada SQL Editor, jalankan pernyataan SQL berikut untuk mendaftarkan Fluss sebagai katalog eksternal:

   CREATE EXTERNAL CATALOG <catalog_name>
   PROPERTIES (
       'type' = 'fluss',
       'bootstrap.servers' = '<Fluss cluster bootstrap address>',
       'fluss.option.client.security.protocol' = 'SASL',
       'fluss.option.client.security.sasl.mechanism' = 'PLAIN',
       'fluss.option.client.security.sasl.username' = '<Fluss username>',
       'fluss.option.client.security.sasl.password' = '<Fluss password>'
   );

Deskripsi parameter

Setelah katalog dibuat, Anda dapat langsung menjalankan kueri menggunakan nama tiga bagian (Catalog.Database.Table) tanpa perlu mengganti konteks. Nama database Fluss default adalah fluss:

SELECT * FROM <catalog_name>.<database_name>.<table_name>;

Anda juga dapat beralih ke katalog Fluss sebelum menjalankan kueri:

-- Switch catalog and database
SET CATALOG <catalog_name>;
USE <db_name>;

-- Query the target table
SELECT COUNT(*) FROM <table_name> LIMIT 10;

Atau, lakukan dalam satu langkah:

USE <catalog_name>.<db_name>;

Langkah 2: Tentukan mode kueri menggunakan sufiks

Untuk tabel Fluss yang telah mengaktifkan integrasi lakehouse-stream (table.datalake.enabled = 'true' diatur saat pembuatan tabel), datanya tersimpan baik di kluster Fluss (data panas real-time) maupun di data lake Paimon (data historis yang telah ditiering). StarRocks mendukung penentuan mode kueri dengan menambahkan sufiks ke nama tabel, sehingga memungkinkan akses fleksibel ke data di lapisan berbeda. Mode kueri tersebut adalah:

Sebagai contoh, pertimbangkan tabel pesanan penjualan online order, yang mencatat data pesanan pengguna secara real-time (termasuk bidang seperti ID pesanan, jumlah pesanan, waktu pesanan, dan wilayah) dan telah mengaktifkan integrasi lakehouse-stream. Contoh berikut menunjukkan skenario bisnis khas untuk ketiga mode kueri tersebut:

-- Scenario 1: Real-time monitoring of today's order count and amount ($rt, second-level latency)
SELECT COUNT(*) AS order_cnt, SUM(order_amount) AS total_amount
FROM `fluss_catalog`.`fluss`.order$rt
WHERE order_time >= CURRENT_DATE;

-- Scenario 2: Calculate week-over-week sales growth rate ($lake, query Paimon historical data)
SELECT
    this_week AS this_week_sales,
    last_week AS last_week_sales,
    ROUND((this_week - last_week) * 100.0 / NULLIF(last_week, 0), 2) AS wow_growth_rate
FROM (
    SELECT
        SUM(CASE WHEN YEARWEEK(order_time) = YEARWEEK(CURRENT_DATE)     THEN order_amount ELSE 0 END) AS this_week,
        SUM(CASE WHEN YEARWEEK(order_time) = YEARWEEK(CURRENT_DATE) - 1 THEN order_amount ELSE 0 END) AS last_week
    FROM `fluss_catalog`.`fluss`.order$lake
) t;

-- Scenario 3: Query total order amount by region in the past 7 days (no suffix, Union Read merges hot and cold data)
SELECT region, SUM(order_amount) AS total_amount
FROM `fluss_catalog`.`fluss`.order
WHERE order_time >= CURRENT_DATE - INTERVAL '7' DAY
GROUP BY region
ORDER BY total_amount DESC;

Lihat katalog yang telah dibuat

-- View all catalogs
SHOW CATALOGS;

-- View the CREATE statement of a specified catalog
SHOW CREATE CATALOG <catalog_name>;

Lihat skema tabel Fluss

DESCRIBE <catalog_name>.<database_name>.<table_name>;

Lihat daftar database Fluss

SHOW DATABASES FROM <catalog_name>;

Hapus katalog Fluss

Menghapus katalog hanya menghapus pemetaan di StarRocks. Data aktual di Fluss tidak terpengaruh.

DROP CATALOG <catalog_name>;