Belajar Apache Flink - Complex Event Processing (CEP)

#Pendahuluan

Sejauh ini kita memproses event satu per satu atau mengagregasinya dalam window. Episode 10 ini mengubah cara pandang: bagaimana jika yang kita cari adalah urutan kejadian — transaksi mencurigakan yang diikuti transaksi lain dalam satu menit, atau serangkaian log yang menandakan kegagalan berantai? Inilah ranah Complex Event Processing (CEP).

Flink menyediakan library CEP lengkap dengan Pattern API. Kita akan belajar mendefinisikan pattern, mencocokkannya dengan aliran event, menerapkannya pada fraud detection dan anomaly detection, serta menangani timed patterns dan pattern states. Ini salah satu kemampuan yang membuat Flink menonjol di kelasnya.

#Konsep CEP di Flink

#Pattern sebagai Deskripsi Urutan

CEP bekerja dengan mendefinisikan pattern — deskripsi urutan event yang ingin dideteksi. Setiap pattern terdiri dari beberapa state: begin, next, followedBy, dengan condition yang memfilter event mana yang cocok. Ketika urutan event di stream sesuai dengan pattern, Flink menghasilkan sebuah match.

begin("a") → next("b") → followedBy("c") → within(1 menit)

Urutan di atas berarti: event a, kemudian b segera setelahnya, kemudian c di suatu tempat sesudahnya, semuanya dalam rentang satu menit.

#Pattern API

Pattern API diekspresikan sebagai rangkaian method call. Mari kita lihat definisi pattern untuk mendeteksi dua transaksi besar berurutan:

import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.windowing.time.Time;
 
Pattern<Transaction, ?> pattern = Pattern.<Transaction>begin("t1")
    .where(new SimpleCondition<Transaction>() {
        @Override
        public boolean filter(Transaction t) {
            return t.getAmount() > 500000;
        }
    })
    .next("t2")
    .where(new SimpleCondition<Transaction>() {
        @Override
        public boolean filter(Transaction t) {
            return t.getAmount() > 500000;
        }
    })
    .within(Time.minutes(1));

.next("t2") menuntut event t2 muncul langsung setelah t1, sedangkan within(Time.minutes(1)) membatasi keseluruhan pola dalam satu menit. Pahami bahwa next bersifat ketat: tidak ada event lain di antaranya.

#Menggunakan CEP pada Stream

#Membungkus Stream dengan CEP.pattern

Pattern tidak berguna sebelum dipasangkan ke stream. Gunakan CEP.pattern lalu proses setiap match:

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
 
DataStream<Transaction> keyed = transactions.keyBy(Transaction::getAccount);
 
PatternStream<Transaction> patternStream = CEP.pattern(keyed, pattern);
 
DataStream<Alert> alerts = patternStream
    .process(new PatternProcessFunction<Transaction, Alert>() {
        @Override
        public void processMatch(
                Map<String, List<Transaction>> match,
                Context ctx,
                Collector<Alert> out) {
            Transaction t1 = match.get("t1").get(0);
            Transaction t2 = match.get("t2").get(0);
            out.collect(new Alert(t1.getAccount(), t1.getAmount(), t2.getAmount()));
        }
    });

match.get("t1") mengambil semua event yang cocok dengan state t1. Dari sini, logika bisnis kalian (membuat alert, menulis ke sink) tinggal disambung.

#Perbedaan next, followedBy, dan Quantifier

• next — event berikutnya harus langsung berurutan.
• followedBy — event berikutnya muncul setelahnya, boleh diselingi event lain.
• times(2) dan oneOrMore() — mengulang sebuah state beberapa kali.

Pattern.<Event>begin("mulai")
    .where(e -> e.getType().equals("ping"))
    .times(2)
    .next("akhir")
    .where(e -> e.getType().equals("pong"));

#Use Case: Fraud Detection dan Anomaly Detection

#Skema Fraud dengan Ambang Ganda

Kasus klasik: dua transaksi besar dari akun yang sama dalam waktu singkat. Pattern di atas sudah menangkap ini. Perkuat dengan menambahkan kondisi per akun, misalnya membandingkan dengan rata-rata historis yang disimpan di state.

#Anomaly dan Workflow Automation

CEP tidak hanya untuk fraud. Contoh lain:

• Anomaly detection: lonjakan error rate diikuti penurunan health check dalam beberapa menit.
• Workflow automation: sequence checkout login → add_to_cart → checkout yang dibaca untuk konversi.
• Failure cascades: serangkaian service down yang menandakan incident besar.

./bin/flink run -d target/cep-job.jar

Perintah ./bin/flink run -d mengirim job CEP ke cluster. Dalam produksi, hasil match biasanya ditulis ke Kafka atau sistem alerting.

#Timed Patterns dan Pattern States

#Batas Waktu dan Pengelolaan State

Pattern dengan .within() membuat Flink menyimpan state sementara untuk setiap stream key sampai batas waktu tercapai atau pola terpenuhi. Ini berarti CEP memakai memori — pantau ukuran state agar tidak membengkak. Gunakan pattern yang sesempit mungkin dan set timeout dengan TimeoutHandler untuk membersihkan partial match.

DataStream<Alert> alerts = patternStream
    .process(new PatternTimeoutFunction<Transaction, Alert>() {
        @Override
        public Alert onTimeout(
                Map<String, List<Transaction>> partial,
                long ts,
                Context ctx) {
            return new Alert("timeout", partial.size());
        }
    })
    .setParallelism(1);

PatternTimeoutFunction dipanggil ketika sebuah pola tidak selesai dalam batas within — memberi kalian visibilitas atas pola yang menggantung, sekaligus kesempatan membersihkan state.

#Penutup

Episode 10 memperkenalkan CEP Flink: mendefinisikan pattern dengan Pattern API, menggabungkannya ke stream dengan CEP.pattern, memproses match dengan PatternProcessFunction, serta menerapkan pola-pola tersebut pada fraud detection, anomaly detection, dan workflow automation. Kalian juga memahami timed patterns dan manajemen pattern states.

Inti yang harus dibawa pulang:

• CEP mendeteksi urutan event, bukan sekadar event individual.
• next menuntut urutan ketat, followedBy melonggarkannya, times mengulang state.
• within membatasi pola pada rentang waktu dan menjadi batas pembersihan state.
• CEP menyimpan state per key — desain pattern sesempit mungkin.
• Use case utama: fraud detection, anomaly detection, dan workflow automation.

Di episode 11 selanjutnya kita akan membahas job configuration & deployment — menyusun package job JAR dengan dependency shading, mode deployment standalone, YARN, dan Kubernetes, konfigurasi resource JobManager dan TaskManager, serta mengelola lifecycle job dengan CLI dan web UI. Kalian akan mengirim job dari laptop hingga ke cluster produksi.