Realtime Compute for Apache Flink:Checkpoint dan titik simpan

• Mengapa data tidak diperbarui setelah TTL status kedaluwarsa saat mini-batch diaktifkan?

• Cara menghitung waktu mulai checkpoint periodik berikutnya

• Apa perbedaan antara GeminiStateBackend di VVR 8.x dan VVR 6.x?

• Apakah normal jika ukuran checkpoint penuh dan checkpoint inkremental sama?

• Apa yang harus dilakukan jika checkpoint lambat untuk penerapan Python

• Cara memecahkan masalah exception checkpoint

• Error: Anda menggunakan mesin status V4 baru untuk memulihkan data status lama dari checkpoint

• Error: java.lang.NegativeArraySizeException

• Error: org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots.

• Error: Exceeded checkpoint tolerable failure threshold

Kegagalan pembaruan data mini-batch

Status menyimpan hasil komputasi penuh sebelumnya. Jika TTL status kedaluwarsa, status tersebut akan dihapus sehingga hasil akumulasi hilang. Akibatnya, data baru tidak dapat diperbarui berdasarkan hasil mini-batch.

Sebaliknya, jika mini-batch dinonaktifkan, data untuk kunci yang kedaluwarsa akan dihitung ulang dan dikirimkan saat TTL status kedaluwarsa. Hal ini memastikan pembaruan data kontinu, tetapi peningkatan frekuensi pembaruan dapat menimbulkan masalah lain, seperti penundaan pemrosesan.

Oleh karena itu, Anda harus mengonfigurasi pengaturan mini-batch dan TTL sesuai dengan kebutuhan bisnis spesifik Anda.

Menghitung waktu mulai checkpoint berikutnya

Waktu mulai checkpoint berikutnya ditentukan oleh dua parameter: interval checkpoint dan jeda minimum antar checkpoint. Checkpoint baru dipicu ketika kedua kondisi berikut terpenuhi:

• Interval checkpoint: Waktu minimum antara awal satu checkpoint dan awal checkpoint berikutnya, yaitu rentang waktu antara <waktu mulai checkpoint sebelumnya, waktu mulai checkpoint berikutnya>.

• Jeda minimum: Waktu minimum antara akhir satu checkpoint dan awal checkpoint berikutnya, yaitu rentang waktu antara <waktu selesai checkpoint sebelumnya, waktu mulai checkpoint berikutnya>.

Pertimbangkan dua skenario berikut dengan interval checkpoint 3 menit, jeda minimum 3 menit, dan timeout 10 menit.

• Skenario 1: Penerapan berjalan normal, dan setiap checkpoint berhasil.

  Checkpoint pertama dimulai pukul 12:00:00 dan selesai sukses pada 12:00:02. Checkpoint kedua akan dimulai pada 12:03:00.

• Skenario 2: Penerapan berjalan tidak normal, dan checkpoint gagal karena timeout.

  Checkpoint pertama dimulai pukul 12:00:00 dan selesai sukses pada 12:00:02. Checkpoint kedua dimulai pada 12:03:00 tetapi gagal pada 12:13:00 karena timeout. Checkpoint ketiga akan dimulai pada 12:16:00.

Untuk informasi lebih lanjut tentang mengonfigurasi jeda minimum antar checkpoint, lihat Tuning Checkpointing.

GeminiStateBackend di VVR 8.x vs. VVR 6.x

Secara default, Realtime Compute for Apache Flink versi VVR 6.x menggunakan GeminiStateBackend V3, sedangkan VVR 8.x menggunakan V4.

Checkpoint penuh dan inkremental berukuran sama

Jika Anda mengamati bahwa ukuran checkpoint penuh dan checkpoint inkremental sama, Anda harus:

• Memverifikasi bahwa checkpoint inkremental telah dikonfigurasi dan diaktifkan dengan benar.

• Perilaku ini mungkin diharapkan dalam skenario tertentu. Misalnya:

  1. Sebelum data masuk (misalnya, sebelum pukul 18:29), penerapan belum memproses data apa pun. Checkpoint hanya berisi status awal sumber, sehingga secara efektif merupakan checkpoint penuh.

  2. Pada pukul 18:29, satu juta catatan masuk. Jika data ini diproses sepenuhnya dalam interval checkpoint berikutnya (misalnya, 3 menit) dan tidak ada data lain yang masuk, checkpoint inkremental pertama akan berisi seluruh status yang dihasilkan dari catatan tersebut.

  Dalam kasus ini, wajar jika ukuran checkpoint penuh dan checkpoint inkremental pertama sama. Checkpoint inkremental pertama harus mencakup status seluruh data untuk memastikan pemulihan penuh dimungkinkan dari titik tersebut, sehingga fungsinya setara dengan checkpoint penuh.

  Manfaat checkpoint inkremental biasanya mulai terlihat sejak checkpoint kedua dan seterusnya. Dengan input data yang stabil dan tanpa perubahan status besar, checkpoint inkremental berikutnya seharusnya lebih kecil, menunjukkan bahwa sistem hanya mengambil snapshot perubahan status. Jika ukurannya tetap sama, Anda perlu menyelidiki status dan perilaku sistem untuk mengidentifikasi potensi masalah.

Checkpoint lambat pada penerapan Python

• Penyebab

  Fungsi User-Defined (UDF) Python yang berkinerja buruk dapat meningkatkan durasi checkpoint dan menurunkan kinerja penerapan.

• Solusi

  Kurangi ukuran buffer. Di bagian Other Configuration, atur parameter berikut. Untuk petunjuknya, lihat Configure custom deployment parameters.

  python.fn-execution.bundle.size: Nilai default: 100000. Satuan: records.
  python.fn-execution.bundle.time: Nilai default: 1000. Satuan: milliseconds.

  Untuk informasi lebih lanjut tentang parameter ini, lihat Flink Python Configuration.

Memecahkan masalah exception checkpoint

1. Diagnosis Jenis Pengecualian

   Lihat riwayat checkpoint di tab Alarm atau State untuk mengidentifikasi jenis exception, seperti timeout atau kegagalan tulis.

   Pilih tab Overview dan perluas bagian Checkpoint. Tabel menampilkan detail setiap checkpoint, termasuk ID, status, waktu pemicu, durasi, dan ukuran data. Kolom Status menunjukkan apakah setiap checkpoint berhasil atau gagal.

2. Isolasi dan selesaikan masalah

   • Skenario 1: Timeout checkpoint sering terjadi. Periksa tekanan balik pada penerapan. Analisis akar penyebab tekanan balik, identifikasi operator lambat, dan selesaikan masalah dengan menyesuaikan sumber daya atau konfigurasi. Untuk informasi lebih lanjut, lihat How to troubleshoot backpressure issues.

   • Skenario 2: Kegagalan tulis checkpoint. Ikuti langkah-langkah berikut untuk menemukan log TaskManager terkait dan menganalisis akar penyebabnya.

     1. Di halaman Checkpoints pada tab Logs, klik Checkpoints History.

        Di halaman Checkpoints History, Anda dapat melihat detail setiap checkpoint, seperti ID, Status, Acknowledged, Trigger Time, End to End Duration, dan Checkpointed Data Size.

     2. Klik tanda plus (+) di samping checkpoint yang gagal untuk melihat detail operatornya.

     3. Perluas operator yang gagal dan klik ID SubTask untuk menuju log TaskManager yang sesuai.

Error: Memulihkan status lama dengan mesin V4

• Pesan error

  Saat melakukan upgrade dari VVR 6.x ke VVR 8.x, Anda mungkin mengalami error: You are using the new V4 state engine to restore old state data from a checkpoint

• Penyebab

  VVR 6.x dan VVR 8.x menggunakan versi berbeda dari GeminiStateBackend, dan checkpoint-nya tidak kompatibel.

• Solusi

  Anda dapat menggunakan salah satu metode berikut untuk menyelesaikan masalah ini:

  • Buat titik simpan dalam format standar dan mulai penerapan dari status tersebut. Untuk informasi lebih lanjut, lihat Manually create a savepoint dan Start a deployment.

  • Restart penerapan tanpa status.

  • (Tidak disarankan) Terus gunakan versi legacy Gemini. Anda harus mengatur parameter state.backend.gemini.engine.type: STREAMING dan restart penerapan agar perubahan berlaku. Untuk informasi cara mengonfigurasi parameter, lihat How to configure deployment parameters.

  • (Tidak disarankan) Terus gunakan mesin VVR 6.x untuk memulai penerapan.

Error: java.lang.NegativeArraySizeException

• Pesan error

  Penerapan yang menggunakan list state mungkin mengalami exception berikut saat runtime:

  Caused by: java.lang.NegativeArraySizeException
    at com.alibaba.gemini.engine.rm.GUnPooledByteBuffer.newTempBuffer(GUnPooledByteBuffer.java:270)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:85)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:75)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.internalGet(PageStoreImpl.java:428)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:271)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:112)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:118)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:57)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.getOrDefault(GeminiSubKeyedListStateImpl.java:97)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:88)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:47)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:60)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:44)
    at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.onProcessingTime(WindowOperator.java:533)
    at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:289)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1435)

• Penyebab

  Data status untuk satu kunci dalam list state telah melebihi 2 GB. Hal ini dapat terjadi sebagai berikut:

  1. Selama operasi normal, nilai yang ditambahkan ke satu kunci dalam list state digabung melalui proses merge (misalnya, dalam operator window), sehingga data status terus bertambah.

  2. Ketika data status mencapai ukuran tertentu, pertama-tama dapat memicu error Out Of Memory (OOM). Setelah penerapan pulih dari kegagalan, proses merge dapat menyebabkan backend status meminta array byte sementara yang melebihi 2 GB, sehingga menimbulkan exception ini.

  Catatan

  RocksDBStateBackend dapat mengalami masalah serupa, yang mungkin memicu ArrayIndexOutOfBoundsException atau segmentation fault. Untuk informasi lebih lanjut, lihat The EmbeddedRocksDBStateBackend.

• Solusi

  • Jika status besar disebabkan oleh operator window, pertimbangkan untuk mengurangi ukuran window.

  • Jika status besar disebabkan oleh logika penerapan, pertimbangkan untuk mendesain ulang, misalnya dengan membagi kunci.

Error: FlinkKafkaException: Too many ongoing snapshots

• Pesan error

  org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints

• Penyebab

  Error ini terjadi saat menggunakan sink Kafka dan disebabkan oleh beberapa kegagalan checkpoint berturut-turut.

• Solusi

  Untuk mencegah kegagalan akibat timeout, tingkatkan timeout checkpoint dengan menyesuaikan parameter execution.checkpointing.timeout. Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Configure custom deployment parameters.

Error: Exceeded checkpoint tolerable failure threshold

• Pesan error

  org.apache.flink.util.FlinkRuntimeException:Exceeded checkpoint tolerable failure threshold.
    at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:66)

• Penyebab

  Jumlah kegagalan checkpoint yang dapat ditoleransi terlalu rendah, sehingga penerapan memicu failover ketika ambang batas ini terlampaui. Jika parameter ini tidak diatur, nilai default-nya adalah 0, artinya tidak ada kegagalan checkpoint yang ditoleransi.

• Solusi

  Sesuaikan jumlah kegagalan checkpoint yang diizinkan dengan mengatur parameter execution.checkpointing.tolerable-failed-checkpoints: num, di mana num harus berupa bilangan bulat positif atau 0. Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Configure custom deployment parameters.