Realtime Compute for Apache Flink:Fitur tabel

Ikhtisar

• remote storage: Untuk memastikan latensi rendah, data terbaru disimpan di disk lokal. Data yang lebih lama dimigrasikan secara asinkron ke penyimpanan remote guna mengurangi biaya penyimpanan lokal.

• column pruning: Saat kueri dijalankan, hanya kolom yang diperlukan yang dibaca dan kolom yang tidak relevan dilewati, sehingga mengurangi jumlah data yang dibaca.

• predicate pushdown: Server menggunakan statistik kolom untuk melewati blok data yang tidak sesuai dengan filter. Hal ini mengurangi beban I/O jaringan dan deserialisasi.

• log compression: Fitur ini menggunakan mekanisme kompresi Arrow native untuk secara signifikan mengurangi biaya transfer jaringan dan penyimpanan.

Remote storage

Remote storage mengacu pada sistem penyimpanan eksternal yang lebih hemat biaya dan andal, seperti Amazon S3, Hadoop Distributed File System (HDFS), atau Alibaba Cloud OSS. Secara default, Fluss menggunakan Alibaba Cloud OSS sebagai penyimpanan remotenya.

• Fluss menggunakan remote storage untuk tiering data dan toleransi kesalahan.

• Untuk tabel log, remote storage menyimpan segmen log bertingkat.

• Untuk tabel primary key, remote storage menyimpan Snapshot dan segmen data change log.

Column pruning

Column pruning adalah optimisasi yang mengurangi jumlah data yang dibaca dengan hanya membaca kolom yang diperlukan dan melewatkan kolom yang tidak relevan selama kueri. Fluss mendukung column pruning untuk pembacaan aliran, menjadikannya salah satu dari sedikit sistem di industri yang memiliki kemampuan ini untuk skenario pemrosesan aliran real-time.

• Fluss mendukung column pruning baik untuk tabel log maupun change log tabel primary key. Fitur ini mengurangi I/O penyimpanan, menurunkan lalu lintas jaringan, dan mengurangi beban parsing di sisi klien, sehingga secara signifikan meningkatkan performa kueri.

• Baik untuk pemrosesan batch maupun pemrosesan aliran, sistem selalu memproses hanya bidang yang diperlukan, sehingga memaksimalkan pemanfaatan resource.

Saat mesin komputasi, seperti Flink, menjalankan kueri, sistem tersebut menganalisis kolom yang diperlukan dan menginstruksikan Fluss untuk hanya mengembalikan kolom-kolom tersebut.

SELECT id, name FROM log_table WHERE timestamp > '2023-01-01'

Kueri ini hanya membaca kolom id, name, dan timestamp. Fluss tidak membaca kolom lain dalam tabel, seperti address dan status, dari penyimpanan, sehingga menghindari pembacaan yang tidak perlu.

Predicate pushdown

Predicate pushdown adalah optimisasi sisi server yang hanya berlaku untuk tabel log. Server mengevaluasi predikat filter terhadap statistik kolom (min, max, dan nullCount) dari setiap RecordBatch. Server kemudian melewatkan seluruh batch yang tidak mungkin berisi baris yang sesuai, sehingga mengurangi beban I/O jaringan dan deserialisasi.

CREATE TABLE log_table (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'table.statistics.columns' = 'order_id,amount'
);

ALTER TABLE log_table SET ('table.statistics.columns' = 'order_id,amount');

Log compression

Fluss menggunakan mekanisme kompresi Arrow native. Data terkompresi tetap sesuai dengan standar format Arrow. Fluss mengompresi setiap kolom secara independen, sehingga memungkinkan column pruning berfungsi dan tidak memengaruhi performa kueri.

Proses kompresi

• Untuk tabel log, writer di sisi klien mengompresi data dan menuliskannya ke penyimpanan dalam format terkompresi. Saat membaca data tersebut, pemindai log mendekompresinya.

• Untuk change log tabel primary key, server menghasilkan dan mengompresi log tersebut.

Manfaat kompresi

Kompresi secara signifikan mengurangi biaya transfer jaringan dan penyimpanan. Uji benchmark menunjukkan bahwa ZSTD pada level kompresi 3 dapat mencapai rasio kompresi sekitar 5:1 (misalnya, mengompresi data 5 GB menjadi 1 GB). Pengurangan overhead jaringan juga meningkatkan throughput baca dan tulis.

Konfigurasi

Contoh konfigurasi

-- Atur algoritma kompresi ke ZSTD dan level kompresi ke 2
CREATE TABLE log_table (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'table.log.arrow.compression.type' = 'ZSTD',
  'table.log.arrow.compression.zstd.level' = '2'
);