Realtime Compute for Apache Flink：FAQ tentang checkpoints atau savepoints suatu deployment

• Mengapa tidak ada data baru yang diperbarui setelah TTL status berakhir saat miniBatch diaktifkan?

• Bagaimana cara menghitung waktu mulai checkpoint berikutnya?

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

• Apa yang harus dilakukan jika ukuran checkpoint inkremental sama dengan ukuran checkpoint penuh?

• Apa yang harus dilakukan jika checkpoint untuk deployment Python dibuat dengan kecepatan rendah?

• Apa yang harus dilakukan jika terjadi kesalahan selama proses checkpointing?

• Apa yang harus dilakukan jika muncul pesan kesalahan "Anda menggunakan mesin state V4 baru untuk memulihkan data state lama dari checkpoint"?

• Apa yang harus dilakukan jika muncul pesan kesalahan "java.lang.NegativeArraySizeException"?

• Apa yang harus dilakukan jika muncul pesan kesalahan "org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Terlalu banyak snapshot sedang berlangsung."?

• Apa yang harus dilakukan jika muncul pesan kesalahan "Melampaui batas toleransi kegagalan checkpoint"?

Ketika miniBatch diaktifkan, tidak ada data baru yang diperbarui setelah TTL data status berakhir. Mengapa?

Jika miniBatch diaktifkan, data dihitung dalam batch dan disimpan di status. Data status didasarkan pada hasil perhitungan sebelumnya. Jika status dibersihkan karena time to live (TTL) berakhir, hasil perhitungan sebelumnya juga akan hilang. Akibatnya, data status tidak dapat diperbarui berdasarkan data batch.

Jika miniBatch dinonaktifkan, ketika TTL status berakhir, data kunci yang kedaluwarsa dihitung ulang secara kumulatif. Dengan demikian, data status selalu dapat diperbarui. Namun, peningkatan frekuensi pembaruan data dapat menyebabkan masalah seperti penundaan pemrosesan.

Konfigurasikan miniBatch dan TTL sesuai dengan kebutuhan bisnis Anda.

Bagaimana cara menghitung waktu mulai checkpoint berikutnya?

Waktu mulai checkpoint berikutnya dipengaruhi oleh interval saat ini dan interval minimum. Checkpoint berikutnya dipicu ketika kedua kondisi berikut terpenuhi:

• Interval saat ini adalah perbedaan waktu minimum antara <waktu mulai terakhir, waktu mulai berikutnya>.

• Interval minimum adalah perbedaan waktu minimum antara <waktu akhir terakhir, waktu mulai berikutnya>.

Dalam skenario berikut, interval minimum antara percobaan checkpoint adalah 3 menit, interval minimum adalah 3 menit, dan periode timeout adalah 10 menit.

• Skenario 1: Deployment berjalan normal. Setiap checkpoint berhasil.

  Checkpoint pertama dipicu pada 12:00:00 dan berhasil pada 12:00:02. Checkpoint kedua dipicu pada 12:03:00.

• Skenario 2: Deployment abnormal. (Checkpoint gagal atau timeout karena alasan tertentu. Dalam contoh ini, checkpoint timeout.)

  Checkpoint pertama dipicu pada 12:00:00 dan berhasil pada 12:00:02. Checkpoint kedua dipicu pada 12:03:00 tetapi gagal pada 12:13:00 karena timeout. Checkpoint ketiga dipicu pada 12:16:00.

Untuk informasi lebih lanjut tentang pengaturan interval checkpoint minimum, lihat Pengaturan Checkpointing.

Apa perbedaan antara GeminiStateBackend di VVR 8.X dan GeminiStateBackend di VVR 6.X?

Secara default, GeminiStateBackend V3 digunakan di Ververica Runtime (VVR) 6.X dari Realtime Compute for Apache Flink, sedangkan GeminiStateBackend V4 digunakan di VVR 8.X.

Apa yang harus saya lakukan jika ukuran checkpoint inkremental sama dengan ukuran checkpoint penuh?

Jika ukuran checkpoint inkremental sama dengan ukuran checkpoint penuh saat menggunakan Realtime Compute for Apache Flink, lakukan langkah-langkah berikut untuk memeriksa apakah ada masalah:

• Periksa apakah savepoint inkremental dikonfigurasi dengan benar dan berfungsi.

• Periksa apakah deployment berjalan dalam skenario khusus. Dalam beberapa skenario khusus, ukuran checkpoint inkremental diharapkan sama dengan ukuran checkpoint penuh. Contohnya:

  1. Sebelum data diinjeksikan ke deployment pada 18:29, deployment tidak memproses data apa pun. Dalam kasus ini, checkpoint hanya berisi data status awal dari sumber data. Checkpoint tersebut merupakan checkpoint penuh karena tidak ada data status yang ada.

  2. Sebanyak 1.000.000 catatan data diinjeksikan ke deployment pada 18:29. Jika catatan data sepenuhnya diproses dalam interval checkpoint saat ini dan tidak ada data lain yang diinjeksikan selama interval tersebut, checkpoint inkremental pertama yang dihasilkan berisi semua data status dari 1.000.000 catatan data. Interval checkpoint adalah 3 menit.

  Dalam kasus ini, ukuran checkpoint inkremental sama dengan ukuran checkpoint penuh. Checkpoint inkremental pertama harus berisi status data penuh untuk memastikan bahwa seluruh status dapat dipulihkan dari checkpoint inkremental. Oleh karena itu, checkpoint inkremental pertama sebenarnya adalah checkpoint penuh.

  Dalam kebanyakan kasus, jika input data stabil dan tidak ada perubahan status berskala besar, ukuran checkpoint inkremental kedua dan seterusnya berbeda dari ukuran checkpoint penuh. Ini menunjukkan bahwa sistem membuat savepoint hanya untuk data inkremental dari data status seperti yang diharapkan. Jika ukuran checkpoint kedua atau berikutnya sama dengan ukuran checkpoint penuh, periksa status dan perilaku sistem untuk menentukan apakah ada masalah sistem.

Apa yang harus saya lakukan jika checkpoint untuk deployment Python dibuat dengan kecepatan rendah?

• Penyebab

  Operator Python memiliki cache. Saat sistem membuat checkpoint, sistem harus memproses semua data dalam cache. Jika performa fungsi Python yang ditentukan pengguna (UDF) buruk, waktu yang diperlukan untuk membuat checkpoint bertambah. Hal ini memengaruhi eksekusi deployment Python.

• Solusi

  Kurangi jumlah data yang dapat di-cache. Anda dapat menambahkan konfigurasi berikut ke bidang Other Configuration di bagian Parameter tab Konfigurasi. Untuk informasi lebih lanjut, lihat Bagaimana cara mengonfigurasi parameter kustom untuk menjalankan deployment?

  python.fn-execution.bundle.size: Jumlah maksimum elemen yang dapat disertakan dalam bundel. Nilai default: 100000. 
  python.fn-execution.bundle.time: Nilai default: 1000. Unit: milidetik.

  Untuk informasi lebih lanjut tentang parameter, lihat Konfigurasi Python Flink.

Apa yang harus saya lakukan jika terjadi kesalahan selama proses checkpointing suatu deployment?

1. Identifikasi jenis kesalahan

   Anda dapat melihat informasi historis checkpoint di tab Alarm atau State untuk mengidentifikasi jenis kesalahan, seperti timeout checkpointing atau kegagalan penulisan.

   

2. Troubleshoot kesalahan berbagai jenis

   • Skenario 1: Jika timeout checkpointing sering terjadi, periksa apakah deployment memiliki tekanan balik. Jika deployment memiliki tekanan balik, analisis penyebab utama tekanan balik, temukan operator lambat, dan sesuaikan sumber daya atau konfigurasi terkait. Untuk informasi lebih lanjut tentang cara menyelesaikan masalah tekanan balik, lihat Bagaimana cara menyelesaikan masalah tekanan balik?

   • Skenario 2: Jika terjadi kegagalan penulisan selama checkpointing, ikuti langkah-langkah berikut untuk memeriksa log TaskManager dan menyelesaikan kesalahan berdasarkan informasi log.

     1. Di subtab Checkpoints tab Logs deployment, klik Riwayat Checkpoints.

        

     2. Klik tanda plus (+) di sebelah kiri checkpoint abnormal untuk melihat detail tentang operator yang terkait dengan checkpoint abnormal.

     3. Klik tanda plus (+) di sebelah kiri operator abnormal, dan klik nilai di kolom ID subtask abnormal untuk pergi ke TaskManager terkait.

        

Apa yang harus saya lakukan jika muncul pesan kesalahan "You are using the new V4 state engine to restore old state data from a checkpoint"?

• Detail kesalahan

  Saat meningkatkan versi VVR dari 6.X ke 8.X, muncul pesan kesalahan "You are using the new V4 state engine to restore old state data from a checkpoint".

• Penyebab

  Versi GeminiStateBackend yang digunakan oleh VVR 6.X tidak konsisten dengan versi GeminiStateBackend yang digunakan oleh VVR 8.X. Oleh karena itu, checkpoint dari kedua versi tidak kompatibel.

• Solusi

  Gunakan salah satu metode berikut untuk menyelesaikan masalah:

  • Buat savepoint dalam format standar untuk deployment Anda dan mulai deployment dari status savepoint. Untuk informasi lebih lanjut, lihat Buat savepoint secara manual dan Mulai deployment.

  • Mulai ulang deployment tanpa status.

  • (Tidak direkomendasikan) Gunakan GeminiStateBackend V3. Dalam hal ini, Anda harus mengonfigurasi state.backend.gemini.engine.type: STREAMING dan mulai ulang deployment. Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Bagaimana cara mengonfigurasi parameter untuk menjalankan deployment?.

  • (Tidak direkomendasikan) Gunakan mesin VVR 6.X untuk memulai deployment.

Apa yang harus saya lakukan jika muncul pesan kesalahan "java.lang.NegativeArraySizeException"?

• Detail kesalahan

  Jika deployment menggunakan list state, pengecualian berikut terjadi saat deployment berjalan:

  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

  Ukuran data status dari satu kunci dalam list state melebihi 2 GB. Sejumlah besar data status dihasilkan dalam proses berikut:

  1. Saat deployment berjalan sesuai harapan, nilai-nilai yang ditambahkan ke nilai untuk kunci dalam list state digabungkan dalam proses penggabungan. Misalnya, data status terus-menerus terakumulasi untuk deployment yang menggunakan list state dari operator jendela.

  2. Saat ukuran data status terakumulasi hingga ambang batas tertentu, kesalahan kehabisan memori (OOM) dipicu. Setelah deployment pulih dari kegagalan, proses penggabungan list state menyebabkan ukuran larik byte sementara yang diminta oleh backend status melebihi batas. Akibatnya, pengecualian ini terjadi.

  Catatan

  Saat menggunakan RocksDBStateBackend, masalah ini juga dapat terjadi dan pesan kesalahan "ArrayIndexOutOfBoundsException" atau "Segmentation fault" muncul. Untuk informasi lebih lanjut, lihat The EmbeddedRocksDBStateBackend.

• Solusi

  • Jika operator jendela menghasilkan sejumlah besar data status, kami sarankan Anda mengurangi ukuran jendela.

  • Jika logika deployment menghasilkan sejumlah besar data status yang berlebihan, kami sarankan Anda memodifikasi logika deployment. Misalnya, Anda dapat membagi kunci.

Apa yang harus saya lakukan jika muncul pesan kesalahan "org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots."?

• Detail kesalahan

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

• Penyebab

  Kesalahan ini disebabkan oleh beberapa kegagalan checkpoint berturut-turut jika Kafka digunakan sebagai sink.

• Solusi

  Ubah periode timeout checkpoint dengan menyetel parameter execution.checkpointing.timeout untuk memastikan bahwa checkpoint tidak gagal karena timeout. Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Bagaimana cara mengonfigurasi parameter kustom untuk menjalankan deployment?.

Apa yang harus saya lakukan jika muncul pesan kesalahan "Exceeded checkpoint tolerable failure threshold"?

• Detail Kesalahan

  org.apache.flink.util.FlinkRuntimeException:Melampaui batas toleransi kegagalan checkpoint.
    at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:66)

• Penyebab

  Jumlah kegagalan checkpoint yang dapat ditoleransi diatur pada nilai kecil. Deployment memicu failover setelah jumlah kegagalan checkpoint melebihi batas toleransi. Secara default, tidak ada kegagalan checkpoint yang dapat ditoleransi.

• Solusi

  Atur parameter execution.checkpointing.tolerable-failed-checkpoints: num untuk menyesuaikan jumlah kegagalan checkpoint yang dapat ditoleransi untuk deployment. Parameter ini harus diatur ke 0 atau bilangan bulat positif. Jika diatur ke 0, tidak ada pengecualian atau kegagalan checkpoint yang dapat ditoleransi. Untuk informasi lebih lanjut tentang cara mengonfigurasi parameter, lihat Bagaimana cara mengonfigurasi parameter kustom untuk menjalankan deployment?.