Realtime Compute for Apache Flink：Mengembangkan Pemicu Khusus

Apa yang dilakukan pemicu

1. Memantau elemen data yang masuk ke jendela.

2. Menentukan apakah akan memicu pemrosesan berdasarkan kondisi tertentu, seperti waktu atau jumlah peristiwa.

3. Mengembalikan salah satu dari operasi berikut:

   • CONTINUE: Lanjutkan mengumpulkan data.

   • FIRE: Pemicu komputasi dan pertahankan status.

   • FIRE_AND_PURGE: Pemicu komputasi dan bersihkan status.

   • PURGE: Bersihkan status tanpa memicu komputasi.

Pemicu bawaan

Flink menyediakan pemicu bawaan berikut untuk operasi jendela berbasis waktu atau hitungan:

Dalam kasus penggunaan dunia nyata, sering kali kita perlu:

• Menggunakan lebih dari satu kondisi pemicu, seperti memicu jendela setelah 5 peristiwa atau 1 menit.

• Memacu jendela berdasarkan peristiwa tertentu, seperti logout pengguna atau penyelesaian pesanan.

• Mengelola siklus hidup jendela, seperti penembakan awal, penembakan tertunda, dan penembakan berulang.

• Menghindari efek samping perilaku default, seperti jendela yang dipicu ulang oleh data terlambat.

Permintaan kompleks ini tidak dapat dipenuhi oleh pemicu bawaan, sehingga membuat pemicu khusus menjadi diperlukan.

Dalam jendela tumbling waktu-peristiwa selama 1 jam, komputasi jendela dipicu ketika jendela berakhir. Selain itu, komputasi langsung dipicu pada kedatangan peristiwa kelima pengguna dalam jendela tersebut. Pemicu awal ini hanya dipicu sekali per jendela.

Kode contoh:

public class CustomCountTrigger extends Trigger<UserEvent, TimeWindow> {

    // Rekam jumlah elemen setiap kunci dalam jendela
    private final ValueStateDescriptor<Integer> countStateDesc =
            new ValueStateDescriptor<>("count", Integer.class);
    // Rekam apakah komputasi sudah dipicu
    private final ValueStateDescriptor<Boolean> flagStateDesc =
            new ValueStateDescriptor<>("flag", Boolean.class);

    // Panggil onElement() ketika elemen baru masuk ke jendela
    @Override
    public TriggerResult onElement(UserEvent event, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
        // Dapatkan status hitungan kunci saat ini dalam jendela saat ini
        ValueState<Integer> countState = ctx.getPartitionedState(countStateDesc);
        ValueState<Boolean> flagState = ctx.getPartitionedState(flagStateDesc);
        int count = countState.value() == null ? 0 : countState.value();
        boolean flag = flagState.value() == null ? false : flagState.value();

        // Tambahkan hitungan untuk setiap elemen
        count += 1;
        countState.update(count); // Perbarui status
        
        // Ketika hitungan mencapai 5, picu komputasi jendela segera
        if (count >= 5 && !flag) {
            flagState.update(true); // Perbarui status untuk memastikan komputasi tambahan ini hanya dipicu sekali
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext triggerContext) throws Exception {
        // Lewati timer waktu pemrosesan dalam pemicu ini.
        return TriggerResult.CONTINUE;
    }

    // Panggil onEventTime() ketika timer waktu peristiwa yang terdaftar aktif (misalnya, waktu akhir jendela)
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return TriggerResult.FIRE_AND_PURGE; // Pemicu aktif dan jendela dibersihkan
    }

    @Override
    public void clear(TimeWindow timeWindow, TriggerContext ctx) throws Exception {
        // Bersihkan status jendela
        ctx.getPartitionedState(countStateDesc).clear();
        ctx.getPartitionedState(flagStateDesc).clear();
    }
}

Kasus penggunaan 1

Picu komputasi lebih dari sekali sebelum jendela berakhir untuk peringatan.

1. Hapus flagStateDesc untuk memungkinkan komputasi dipicu lebih dari sekali.

2. Atau, tambahkan penanda hitungan untuk mengakhiri jendela hanya setelah sejumlah pemotretan pemicu (seperti untuk peristiwa peringatan).

   ### Setelah menghapus flagStateDesc, 5 rekaman dikeluarkan.  
   Key: 101, Jumlah akses: 5       // count >= 5, memicu komputasi awal dan tidak membersihkan status.
   Key: 101, Jumlah akses: 6       // count >= 5, memicu komputasi awal.
   Key: 101, Jumlah akses: 7      // count >= 5, memicu komputasi awal.
   Key: 101, Jumlah akses: 8      // count >= 5, memicu komputasi awal.
   Key: 101, Jumlah akses: 8      // Memicu komputasi ketika jendela ditutup, dan membersihkan status.

3. Pemicu hanya memutuskan kapan memicu komputasi. Informasi objek aktual diperoleh dari .process, di mana Anda dapat menentukan menghasilkan hasil koleksi yang berbeda berdasarkan jumlah dan kondisi status.

Kasus penggunaan 2

Penutupan jendela memerlukan kemajuan watermark. Bagaimana cara memastikan komputasi jendela tepat waktu jika tidak ada data yang dihasilkan untuk periode waktu yang lama?

• Solusi 1: Gunakan Jendela Waktu Pemrosesan

  Jika Anda tidak memerlukan semantik waktu peristiwa (yaitu, waktu terjadinya peristiwa bukanlah perhatian), Anda dapat menggunakan jendela waktu pemrosesan:

  .keyBy(keySelector)
  .window(TumblingProcessingTimeWindows.of(Time.seconds(60))) // Gunakan waktu pemrosesan
  .process(new MyProcessWindowFunction())

• Solusi 2: Gunakan Metode withIdleness

  Strategi Watermark Flink WatermarkStrategy menyediakan metode withIdleness untuk secara otomatis menandai sumber data sebagai idle setelah periode inaktivitas tertentu, mencegahnya menghalangi pembuatan watermark.

  // Sumber data idle tidak lagi terlibat dalam komputasi watermark minimum, dan tidak akan menghambat kemajuan watermark dari sumber data aktif.
  WatermarkStrategy
      .<Tuple2<Long, String>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
      .withIdleness(Duration.ofMinutes(1));  // Menunjukkan bahwa jika sumber data atau partisi tidak memiliki peristiwa selama 1 menit, itu ditandai sebagai idle

• Solusi 3: Gunakan Generator Watermark Khusus

  Jika Anda perlu menggunakan semantik waktu peristiwa, tentukan generator watermark khusus yang memastikan watermark terus maju meskipun tidak ada data baru yang tiba:

  • Rekam waktu kedatangan peristiwa terbaru dalam metode onEvent().

  • Periksa interval antara waktu saat ini dan waktu terakhir peristiwa diterima dalam metode onPeriodicEmit().

  • Jika interval melebihi ambang batas yang ditetapkan, anggap sumber data sebagai idle, lewati pembuatan watermark atau langsung buat watermark tertentu.

  public class IdleAwareWatermarkGenerator implements WatermarkGenerator<MyEvent> {
      private long lastEventTimestamp = Long.MIN_VALUE;
      private final long maxIdleTimeMs; // Waktu idle maksimum
  
      public IdleAwareWatermarkGenerator(long maxIdleTimeMs) {
          this.maxIdleTimeMs = maxIdleTimeMs;
      }
  
      @Override
      public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
          lastEventTimestamp = Math.max(lastEventTimestamp, eventTimestamp);
      }
  
      @Override
      public void onPeriodicEmit(WatermarkOutput output) {
          long currentTime = System.currentTimeMillis();
          if (lastEventTimestamp == Long.MIN_VALUE || currentTime - lastEventTimestamp > maxIdleTimeMs) {
              // Jika tidak ada peristiwa yang tiba untuk waktu yang lama, jangan keluarkan watermark baru
              return;
          }
          output.emitWatermark(new Watermark(lastEventTimestamp));
      }
  }

  WatermarkStrategy<MyEvent> strategy = WatermarkStrategy
      .forGenerator((ctx) -> new IdleAwareWatermarkGenerator(60_000)) // Tetapkan waktu idle maksimum menjadi 60 detik
      .withTimestampAssigner((event, timestamp) -> event.getEventTime());

• Solusi 4: Daftarkan Timer

  @Override
  public TriggerResult onElement(Event event, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
      ctx.registerEventTimeTimer(window.maxTimestamp()); // Daftarkan timer waktu peristiwa
      ctx.registerProcessingTimeTimer(window.maxTimestamp() + 1000); // Toleransi kesalahan: pastikan timer dipicu meskipun tidak ada peristiwa
      return TriggerResult.CONTINUE;
  }
  
  @Override
  public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
      return TriggerResult.FIRE_AND_PURGE; // Pemicuan paksa dan pembersihan
  }

• Solusi 5: Kirim Pesan Denyut Jantung dari Sistem Eksternal

  Gunakan Kafka atau alur kerja untuk mengirim pesan denyut jantung secara teratur ke hilir untuk memicu penutupan jendela.

Ringkasan