Baca data

Topik ini menjelaskan cara membaca data dari Fluss dalam mode stream dan mode batch.

Mode stream

Fluss mendukung beberapa mode konsumsi. Untuk tabel primary key, mode konsumsi default adalah full, yang pertama kali mengonsumsi data lengkap lalu dilanjutkan dengan data inkremental.

Masuk ke Konsol Realtime Compute for Apache Flink.
Pada kolom Actions ruang kerja target, klik Console.
Di panel navigasi sebelah kiri, klik Development > ETL.
Klik +, lalu klik New Blank Stream Draft. Masukkan Name, pilih Engine Version, lalu klik Create.

Tulis kode pekerjaan.

SELECT * FROM `fluss-catalog`.`my_db`.`my_table`;

# Gunakan mode konsumsi earliest
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

Mode batch

Baik tabel primary key maupun tabel log mendukung pembacaan LIMIT untuk segera mengembalikan jumlah hasil tertentu, sehingga mempermudah pratinjau data.
Baik tabel primary key maupun tabel log (versi 0.9-ali-1.0 dan yang lebih baru) mendukung kueri agregat COUNT(*).
Tabel primary key juga mendukung point lookup dengan menggunakan pencocokan eksak pada primary key dan index key.

Masuk ke Konsol Realtime Compute for Apache Flink.
Pada kolom Actions ruang kerja target, klik Console.
Di panel navigasi sebelah kiri, klik Development > ETL.
Klik +, lalu klik New Blank Batch Draft. Masukkan Name, pilih Engine Version dan SQL Dialect, lalu klik Create.

Tulis kode pekerjaan.

# Pembacaan LIMIT
SELECT * FROM `fluss-catalog`.`my_db`.`my_table` LIMIT 10;

# Point lookup (hanya untuk tabel primary key)
SELECT * FROM `fluss-catalog`.`my_db`.`my_pk_tbl` WHERE shop_id = 10000 and user_id = 123456;

# Kueri agregat
SELECT count(*) FROM `fluss-catalog`.`my_db`.`my_log_table`;

Pengoptimalan kueri

Predicate pushdown

Untuk kondisi filter yang ditulis dalam klausa SELECT ... WHERE, Flink secara otomatis mendorong predikat yang sesuai ke server Fluss. Server kemudian mengevaluasi predikat tersebut terhadap statistik kolom setiap RecordBatch, sehingga dapat melewati blok data yang tidak cocok serta mengurangi overhead transfer jaringan dan deserialisasi.

Operator filter

Operator perbandingan: =, <>, >, >=, <, dan <=
IN (...)
IS NULL, IS NOT NULL
BETWEEN ... AND ...
LIKE: Mendukung pencocokan awalan ('abc%'), pencocokan akhiran ('%abc'), dan pencocokan berisi ('%abc%').
Operator logika: AND, OR

Batasan

Hanya berlaku untuk tabel log.
Semua kolom yang dirujuk dalam kondisi filter harus memiliki statistik kolom yang diaktifkan (table.statistics.columns).
Jika statistik tidak diaktifkan untuk kolom yang dirujuk, kondisi filter tidak akan didorong ke bawah.
Filtering kasar ini berdasarkan statistik tingkat batch dan dapat menghasilkan false positive (mengembalikan data tambahan). Hal ini tidak memengaruhi semantik hasil kueri.

Contoh

Langkah 1: Buat tabel dan aktifkan statistik kolom

CREATE TABLE sensor_log (
  sensor_id BIGINT,
  temperature DOUBLE,
  humidity DOUBLE,
  event_time TIMESTAMP(3)
) WITH (
  'table.statistics.columns' = 'sensor_id,temperature'
);

Langkah 2: Kueri data menggunakan klausa WHERE

SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

Langkah 3: Verifikasi predicate pushdown menggunakan EXPLAIN

EXPLAIN SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

Dalam output EXPLAIN, klausa filter=[...] pada node TableSourceScan menunjukkan bahwa kondisi filter telah didorong ke server Fluss.

Catatan

Merupakan perilaku yang diharapkan jika node Calc dalam output EXPLAIN tetap menyimpan kondisi filter. Klien Flink melakukan penyaringan ulang terhadap data yang diterima sebagai tindakan pengaman.

Untuk detail lebih lanjut mengenai prinsip dan konfigurasi predicate pushdown, lihat Table Properties.

Kueri data

Catatan

Baik tabel primary key maupun tabel log mendukung pembacaan LIMIT untuk segera mengembalikan jumlah hasil tertentu, sehingga mempermudah pratinjau data.
Baik tabel primary key maupun tabel log (versi 0.9-ali-1.0 dan yang lebih baru) mendukung kueri agregat COUNT(*).
Tabel primary key juga mendukung point lookup dengan menggunakan pencocokan eksak pada primary key dan index key.

Masuk ke Konsol Realtime Compute for Apache Flink.
Pada kolom Actions ruang kerja target, klik Console.

Di panel navigasi sebelah kiri, klik Scripts > New Script.

SELECT * FROM `fluss-catalog`.`my_db`.`customer` 
WHERE c_custkey = 1;

Pilih blok kode, lalu klik Run untuk melihat hasilnya.

Mode konsumsi

Mode konsumsi	Tabel kunci utama	Tabel log	Kasus penggunaan
full (Default)	Pertama kali mengonsumsi data lengkap, lalu dilanjutkan dengan data inkremental.	Mengonsumsi data dari offset paling awal.	Ideal untuk skenario yang memerlukan pemrosesan seluruh data, seperti analisis data lengkap.
earliest	Mengonsumsi data mulai dari offset binlog paling awal.	Mengonsumsi data dari offset paling awal.	Ideal untuk skenario yang memerlukan pemrosesan data historis yang berubah, seperti pemuatan data awal.
latest	Mengonsumsi data mulai dari offset binlog terbaru.	Mengonsumsi data dari offset terbaru.	Ideal untuk skenario yang hanya membutuhkan data terbaru, seperti pemantauan waktu nyata.
timestamp	Mengonsumsi data binlog mulai dari waktu tertentu yang ditentukan oleh opsi `scan.startup.timestamp`.	Mengonsumsi data mulai dari waktu tertentu yang ditentukan oleh opsi `scan.startup.timestamp`.	Ideal untuk pemulihan data atau pelacakan balik berdasarkan titik waktu tertentu.

Contoh SQL

// mode full (default)
SELECT * FROM my_table ;

// mode konsumsi earliest
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

// mode konsumsi latest
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'latest') */;

// Gunakan timestamp
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '1678883047356') */;

// Gunakan string waktu
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '2023-12-09 23:09:12') */;