Realtime Compute for Apache Flink：Agregasi data real-time dengan Flink

Latar belakang dan tantangan

Sistem agregasi data real-time tradisional menghadapi beberapa tantangan dalam kasus penggunaan dunia nyata, termasuk:

• Menangani Data Terlambat: Dalam pemrosesan real-time terdistribusi, data sering tiba tidak berurutan karena latensi jaringan, jitter sistem, atau fluktuasi upstream. Tanpa mekanisme penanganan data terlambat yang tepat, seperti watermark, rollback status, atau pengolahan ulang berbasis jendela, data historis dapat ditimpa oleh hasil yang salah. Ini menyebabkan bias statistik, memengaruhi akurasi pemantauan dan keandalan pengambilan keputusan.

• Manajemen Status yang Rumit: Secara tradisional, menangani data berdimensi tinggi atau kesenjangan data dapat menyebabkan ukuran status bertambah secara eksponensial. Ukuran status yang besar mengonsumsi sumber daya memori yang berlebihan, memperlambat checkpointing, meningkatkan waktu pemulihan, merusak stabilitas sistem, dan meningkatkan risiko kegagalan pekerjaan.

• Menyeimbangkan Konsumsi Sumber Daya dan Kinerja: Agregasi real-time harus mempertimbangkan beberapa faktor, termasuk overhead komputasi, penggunaan penyimpanan, dan akurasi hasil. Terlalu bergantung pada status dalam memori meningkatkan biaya sumber daya, sementara pembacaan dan penulisan sering ke penyimpanan eksternal dapat menciptakan hambatan I/O yang memengaruhi throughput dan latensi.

Perbandingan solusi

Agregasi stateful tradisional

Solusi ini menggunakan mekanisme bawaan Flink untuk mempertahankan status dalam memori guna melakukan agregasi data real-time.

Potongan SQL di bawah ini mengagregasi aliran log (view_source) secara real-time. Ini menghitung tampilan halaman (pv) dan klik berdasarkan waktu (ts) dan kluster, lalu menulis hasilnya ke sink_table.

INSERT INTO sink_table
SELECT 
  ts, cluster,
  SUM(pv) as pv,
  SUM(click) as click 
FROM view_source 
GROUP BY ts, cluster;

Deskripsi solusi

Ini adalah solusi dasar untuk agregasi real-time. Menggunakan kemampuan agregasi stateful Flink, memungkinkan Anda mengembangkan tugas agregasi data menggunakan sintaks SQL yang sudah dikenal. Saat data mengalir melalui operator, mesin menemukan dan memperbarui status berdasarkan kunci yang telah ditentukan sebelumnya, lalu menulis hasil agregasi ke penyimpanan hilir.

Manfaat dan keterbatasan

Solusi ini mudah diimplementasikan dan memberikan kesegaran data tinggi. Namun, memiliki keterbatasan yang jelas:

• Data Terlambat Merusak Agregasi Setelah Kedaluwarsa Status: Pertimbangkan jendela waktu yang telah di-aggregasi dengan benar menjadi hasil akhir (misalnya, pv=999). Ketika status untuk jendela ini kedaluwarsa (misalnya, pada pukul 09:30), event terlambat apa pun dengan timestamp dalam jendela tersebut tidak akan menemukan status yang ada. Sebagai gantinya, Flink mungkin menginisialisasi status baru dan mengeluarkan hasil yang tidak lengkap (misalnya, pv=1), yang kemudian menimpa hasil sebelumnya yang benar di sink.

• Pembengkakan Status Merusak Kinerja: Konkurensi tinggi pada dataset besar dapat menyebabkan hotspot status agregasi. Ini mengarah pada ekspansi status cepat, konsumsi memori tinggi, dan perlambatan signifikan dalam checkpointing, memengaruhi stabilitas sistem dan waktu pemulihan.

• Pemulihan Kesalahan yang Lama: Selama pemulihan kesalahan sistem, seluruh status harus dimuat dari checkpoint. Semakin besar status, semakin lama waktu pemulihan, yang memengaruhi ketersediaan sistem.

Agregasi inkremental tanpa status dengan UDAF

Solusi ini mendorong manajemen status kompleks ke sistem penyimpanan. Mesin pemrosesan aliran melakukan perhitungan inkremental tanpa status, dan bergantung pada penyimpanan hilir untuk menghitung hasil agregasi akhir.

Fungsi agregat yang didefinisikan pengguna (UDAF) berikut mengimplementasikan agregasi tanpa status. Ini melakukan perhitungan inkremental pada data dalam batch atau jendela saat ini.

public class LongSumAggUDAF extends AggregateFunction<Long, LongAccumulator> {
    @Override
    public LongAccumulator createAccumulator() {
        return new LongAccumulator();
    }
    
    public void accumulate(LongAccumulator acc, Long value) {
        acc.add(value); // Mengakumulasi data dalam micro-batch saat ini
    }
    
    @Override
    public Long getValue(LongAccumulator acc) {
        return acc.getValue();
    }
}

Deskripsi solusi

• UDAF: Tidak seperti fungsi agregat standar, UDAF melakukan perhitungan inkremental hanya pada data dalam batch atau jendela saat ini di tingkat operator. Ini tidak mempertahankan status historis di seluruh batch, fokus hanya pada nilai micro-batch saat ini.

• Agregasi Inkremental melalui "Read-Compute-Write": Konektor sink pertama-tama membaca nilai agregasi terakhir yang diketahui untuk kunci utama tertentu dari penyimpanan. Kemudian menggabungkan nilai historis dengan hasil inkremental dari batch data saat ini. Terakhir, ia menulis hasil agregasi baru kembali ke penyimpanan. Pola "read-compute-write" ini memastikan agregasi yang benar, bahkan jika pekerjaan dimulai ulang atau data terlambat tiba.

Manfaat dan keterbatasan

Solusi ini menangani data terlambat dengan sempurna. Ini mencegah hasil ditimpa secara salah karena kedaluwarsa status. Selain itu, karena mesin tidak lagi mempertahankan ukuran status besar, efisiensi memori dan kinerja checkpointing meningkat pesat. Namun, kelemahannya adalah solusi ini memerlukan operasi baca tambahan untuk setiap batch atau jendela, yang meningkatkan beban pada sistem penyimpanan dan latensi pemrosesan.

Tabel agregasi perantara

Solusi ini memperkenalkan tabel data lake, seperti tabel Paimon, untuk menyimpan hasil agregasi. Seperti yang ditunjukkan dalam ilustrasi, pipeline agregasi bekerja sebagai berikut: Flink melakukan agregasi tanpa status pada data mentah, menggabungkan hasilnya di tabel lake, dan mengirimkannya ke hilir.

Kode berikut membuat tabel Paimon. Ketika kunci utama cocok, kolom pv dan click di-aggregasi menggunakan sum.

-- Buat tabel agregasi Paimon
CREATE TABLE paimon_agg (
    ts TIMESTAMP(3),
    cluster STRING,
    pv BIGINT,
    click BIGINT,
    PRIMARY KEY (ts, cluster) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.pv.aggregate-function' = 'sum',
    'fields.click.aggregate-function' = 'sum'
);

Deskripsi solusi

Solusi ini menggunakan mekanisme penggabungan data konektor Paimon untuk mengagregasi data. Setel 'merge-engine' = 'aggregation' dan tentukan fungsi agregat, dan konektor Paimon secara otomatis menggabungkan hasil baru dengan hasil historis untuk kolom tertentu. Alur kerja meliputi:

1. Proses Aliran Data: Melakukan agregasi tanpa status dan menulis hasilnya ke tabel Paimon.

2. Gabungkan Data: Menggabungkan hasil agregat dan mengelola versi.

3. Kirim Hasilnya: Mengirim hasilnya ke hilir dalam batch dan interval reguler.

Manfaat

• Mengurangi Tekanan pada Penyimpanan Hilir: Sinkronisasi batch secara signifikan mengurangi beban tulis real-time, meningkatkan stabilitas penyimpanan hilir.

• Logika Agregasi Disederhanakan: Pipeline ini menghilangkan kebutuhan untuk "read-compute-write". Fokus pada ingest data dan pemrosesan inkremental.

• Meningkatkan Fleksibilitas dan Kemudahan Pemeliharaan: Tugas setiap komponen didefinisikan dengan jelas. Selain itu, tabel Paimon terintegrasi mulus dengan berbagai sistem hilir.

• Manajemen Versi dan Pencarian Snapshot Historis

Pemilihan solusi dan pertimbangan

Memilih solusi melibatkan trade-off antara biaya, efisiensi, kompleksitas, skala data, dan kemampuan teknis. Berikut adalah ringkasan manfaat dan keterbatasan setiap solusi:

• Agregasi Stateful: Mudah diimplementasikan dan menawarkan kesegaran tinggi. Tetapi sulit menangani data terlambat dan mengelola status secara efektif.

• Agregasi Inkremental Tanpa Status: Menangani data terlambat secara efektif, menawarkan overhead O&M rendah, dan berlaku luas. Namun, ini memerlukan definisi UDAF yang sesuai dengan logika agregasi Anda.

• Tabel Agregasi Perantara: Mampu menangani dataset besar sambil mengurangi beban pada sistem hilir. Tetapi lebih kompleks untuk diimplementasikan, memerlukan setup dan pemeliharaan yang signifikan.

Catatan bahwa overhead pengembangan dan O&M sering kali berbanding terbalik. Saat memilih solusi, seimbangkan biaya, efisiensi, kompleksitas, skala data, dan kemampuan teknis.