Realtime Compute for Apache Flink:FAQ tentang validitas data

Mengapa saya tidak mendapatkan output di tabel sink?

• Deskripsi

  Setelah pekerjaan dimulai, tidak ada data yang muncul di tabel sink.

• Perbaiki Masalah

  

  1. Periksa apakah terjadi failover.

     • Pemecahan Masalah

       Analisis penyebab failover berdasarkan pesan kesalahan.

     • Solusi

       Selesaikan penyebabnya agar pekerjaan dapat berjalan sesuai harapan.

  2. Verifikasi bahwa data telah masuk ke Realtime Compute for Apache Flink.

     • Pemecahan Masalah

       Jika tidak terjadi failover tetapi latensi data sangat tinggi, lihat nilai metrik numRecordsInOfSource pada halaman pemantauan dan peringatan untuk memeriksa apakah setiap sumber memiliki data masukan.

     • Solusi

       Periksa tabel sumber untuk memastikan data dikirim ke Realtime Compute for Apache Flink.

  3. Periksa apakah data telah difilter oleh operator.

     Tambahkan pipeline.operator-chaining: 'false' ke bidang Other Configuration. Untuk instruksi lebih lanjut, lihat Bagaimana cara mengonfigurasi parameter kustom untuk menjalankan deployment? Pisahkan operator dan amati input (Bytes Received) dan output (Bytes Sent) dari setiap operator untuk menentukan yang bermasalah. Jika operator memiliki input tetapi tidak ada output, data difilter pada operator ini. Masalah ini sering disebabkan oleh salah satu operator ini: JOIN, WINDOW, dan WHERE.

  4. Periksa apakah data di database downstream di-cache berdasarkan mekanisme caching default.

     Solusi: Sesuaikan ukuran batch penyimpanan downstream.

     Penting

     Menggunakan ukuran batch yang terlalu kecil dapat membebani database downstream dan menyebabkan hambatan kinerja. Misalnya, ukuran batch 1 berarti Flink mengirim permintaan untuk setiap rekaman yang diproses, yang memberi beban pada database, terutama dengan volume data yang tinggi.

  5. Periksa apakah database ApsaraDB RDS downstream mengalami deadlock.

     Solusi: Lihat Apa yang harus saya lakukan jika terjadi deadlock saat data ditulis ke database MySQL menggunakan konektor ApsaraDB RDS for MySQL atau konektor TDDL?

Bagaimana cara menyelesaikan masalah pembacaan sumber Flink?

Jika Realtime Compute for Apache Flink tidak dapat membaca data dari sumber, ikuti langkah-langkah berikut:

• Periksa konektivitas jaringan.

  Secara default, Realtime Compute for Apache Flink hanya dapat mengakses layanan dalam wilayah dan virtual private cloud (VPC)-nya sendiri. Untuk mengakses sumber daya lain, lihat topik-topik berikut:

  • Komunikasi lintas-VPC: Bagaimana Realtime Compute for Apache Flink mengakses layanan lintas VPC?

  • Komunikasi melalui Internet: Bagaimana Realtime Compute for Apache Flink mengakses Internet?

• Periksa konfigurasi daftar putih layanan upstream.

  Untuk membaca data dari layanan seperti Kafka dan Elasticsearch, tambahkan Flink ke daftar putih mereka sebagai berikut:

  1. Dapatkan blok CIDR dari vSwitch ruang kerja Flink Anda.

     Untuk informasi lebih lanjut tentang cara mendapatkan blok CIDR, lihat Bagaimana cara mengonfigurasi daftar putih?

  2. Konfigurasikan daftar putih untuk layanan upstream.

     Untuk informasi lebih lanjut, lihat bagian "Prasyarat" dari dokumen konektor masing-masing, seperti Konektor Kafka.

• Periksa konsistensi tipe bidang, urutan bidang, dan huruf besar/kecil nama bidang antara tabel Flink dan tabel fisik.

  Untuk memastikan konsistensi antara tabel Flink dan tabel fisik, ikuti panduan berikut saat menulis pernyataan DDL tabel Flink:

  • Urutan bidang: Replikasi urutan bidang tepat dari tabel fisik.

  • Huruf besar/kecil nama bidang: Gunakan huruf besar/kecil identik untuk nama bidang seperti di tabel fisik.

  • Tipe bidang: Flink dan layanan eksternal mungkin mendukung tipe data yang berbeda. Tipe bidang tabel Flink harus menjadi tipe yang dipetakan setara dari bidang tabel fisik. Untuk detail tentang pemetaan tipe antara Flink dan layanan eksternal, lihat bagian "Pemetaan tipe data" dari dokumen konektor masing-masing, seperti Konektor Layanan Log Sederhana.

• Periksa apakah file Taskmanager.log tabel sumber berisi pesan kesalahan.

  Jika ada pengecualian yang dilaporkan, selesaikan kesalahan berdasarkan pesan kesalahan. Untuk melihat file Taskmanager.log tabel sumber, lakukan langkah-langkah berikut:

  1. Di panel navigasi kiri konsol pengembangan, buka O&M > Deployment.

  2. Klik nama deployment target.

  3. Klik tab Status lalu klik vertex yang mewakili sumber dalam DAG.

  4. Di panel kanan, klik tab SubTasks.

  5. Di kolom More, klik ikon dan pilih Open TaskManager Log Page.

  6. Di tab Logs, lihat informasi log.

     Cari pesan paling awal yang berisi "Caused by". Biasanya ini menunjuk ke penyebab utama pengecualian. Kemudian, Anda dapat menyelesaikan pengecualian berdasarkan pesan tersebut.

  Bagaimana cara menyelesaikan masalah tidak ada output di sistem downstream?

  Selesaikan masalah berikut untuk mengatasi tidak adanya output:

  • Periksa konektivitas jaringan.

    Secara default, Realtime Compute for Apache Flink hanya dapat mengakses layanan dalam wilayah dan virtual private cloud (VPC)-nya sendiri. Untuk akses ke sumber daya lain, lihat topik berikut:

    • Komunikasi lintas-VPC: Bagaimana Realtime Compute for Apache Flink mengakses layanan lintas VPC?

    • Komunikasi melalui Internet: Bagaimana Realtime Compute for Apache Flink mengakses Internet?

  • Periksa konfigurasi daftar putih sistem downstream.

    Untuk menulis data ke layanan seperti ApsaraDB RDS for MySQL, Kafka, Elasticsearch, AnalyticDB for MySQL 3.0, Apache HBase, Redis, dan ClickHouse, tambahkan Flink ke daftar putih mereka:

    1. Dapatkan blok CIDR dari vSwitch ruang kerja Flink Anda.

       Untuk informasi lebih lanjut tentang cara mendapatkan blok CIDR, lihat Bagaimana cara mengonfigurasi daftar putih?

    2. Konfigurasikan daftar putih untuk layanan downstream.

       Untuk informasi lebih lanjut, lihat bagian "Prasyarat" dari dokumen konektor masing-masing, seperti Konektor ApsaraDB RDS for MySQL.

  • Periksa konsistensi tipe bidang, urutan bidang, dan huruf besar/kecil nama bidang antara tabel Flink dan tabel fisik.

    Untuk memastikan konsistensi antara tabel Flink dan tabel fisik, ikuti panduan berikut saat menulis pernyataan DDL tabel Flink:

    • Urutan bidang: Replikasi urutan bidang tepat dari tabel fisik.

    • Huruf besar/kecil nama bidang: Gunakan huruf besar/kecil identik dengan tabel fisik.

    • Tipe bidang: Tipe bidang tabel Flink harus menjadi tipe yang dipetakan setara dari bidang tabel fisik. Untuk detail pemetaan tipe antara Flink dan layanan eksternal, lihat bagian "Pemetaan tipe data" dari dokumen konektor masing-masing, seperti Konektor Layanan Log Sederhana.

  • Periksa apakah data difilter oleh operator perantara, seperti WHERE, JOIN, atau WINDOW.

    Periksa input dan output dari setiap vertex dalam DAG pekerjaan. Misalnya, jika vertex WHERE memiliki input 5 dan output 0, ini menunjukkan bahwa data telah difilter oleh operator WHERE.

  • Periksa apakah nilai default opsi konektor spesifik-sink untuk sistem downstream terlalu besar.

    Saat volume input rendah, nilai default yang terlalu tinggi dari opsi sink tertentu dapat mencegah data tersiram ke sistem downstream karena buffer tidak mencapai ambang keluaran default. Untuk menyelesaikan masalah ini, konfigurasikan opsi terkait dengan nilai yang lebih kecil sesuai kebutuhan:

    Opsi	Deskripsi	Layanan downstream terkait
    batchSize	Ukuran data yang ditulis sekaligus.	DataHub Tablestore MongoDB ApsaraDB RDS for MySQL AnalyticDB for MySQL V3.0 ClickHouse TSDB for InfluxDB
    batchCount	Jumlah maksimum rekaman data yang ditulis sekaligus.	DataHub
    flushIntervalMs	Interval waktu operasi flush dilakukan di buffer penulis di MaxCompute Tunnel.	MaxCompute
    sink.buffer-flush.max-size	Ukuran data dalam byte yang di-cache di memori sebelum data ditulis ke database ApsaraDB for HBase. Nilai parameter ini yang lebih besar meningkatkan kinerja penulisan ApsaraDB for HBase tetapi memperpanjang latensi penulisan dan mengonsumsi lebih banyak memori.	ApsaraDB for HBase
    sink.buffer-flush.max-rows	Jumlah rekaman data yang di-cache di memori sebelum data ditulis ke database ApsaraDB for HBase. Nilai parameter ini yang lebih besar meningkatkan kinerja penulisan ApsaraDB for HBase tetapi memperpanjang latensi penulisan dan mengonsumsi lebih banyak memori.	ApsaraDB for HBase
    sink.buffer-flush.interval	Interval waktu data yang di-cache ditulis ke database ApsaraDB for HBase. Parameter ini mengontrol latensi penulisan data ke database ApsaraDB for HBase.	ApsaraDB for HBase
    jdbcWriteBatchSize	Jumlah maksimum baris data yang dapat diproses oleh node sink streaming Hologres sekaligus saat driver JDBC digunakan.	Hologres

  • Periksa apakah data tidak dapat dikeluarkan karena data tidak berurutan dalam operasi berbasis jendela.

    Misalnya, jika timestamp rekaman awal adalah 2100 dan watermark juga 2100, sistem mengasumsikan bahwa data sebelum 2100 telah diproses. Rekaman berikutnya dengan timestamp seperti 2021 dibuang karena kurang dari watermark 2100. Jendela event-time saat ini tidak dapat ditutup dan tidak ada hasil yang dihitung sampai rekaman dengan timestamp lebih lama dari 2100 tiba.

    Untuk memverifikasi data tidak berurutan di sumber, gunakan tabel sink cetak atau periksa log Log4j. Untuk informasi lebih lanjut, lihat Buat tabel hasil cetak dan Konfigurasikan parameter untuk mengekspor log deployment. Jika ada rekaman tidak berurutan, filter mereka atau izinkan rekaman terlambat diproses.

  • Verifikasi semua operator sumber paralel memiliki input.

    Jika subtask sumber tidak menerima input, watermark-nya tetap pada nilai default: 1970-01-01T00:00:00Z. Ini menjadi watermark keseluruhan operator sumber, sehingga jendela gagal ditutup dan mencegah output data.

    Untuk menyelesaikan masalah, periksa DAG pekerjaan dan pastikan semua subtask sumber menerima input. Jika ada vertex yang tidak menerima input, kurangi paralelisme pekerjaan agar sesuai dengan jumlah shard tabel upstream untuk memastikan semua subtask menerima input.

  • Konfirmasi semua partisi Kafka berisi data.

    Partisi Kafka kosong dapat mencegah pembuatan watermark. Untuk informasi lebih lanjut dan solusi, lihat Mengapa tidak ada output data yang dikembalikan setelah data dari tabel sumber Kafka dihitung menggunakan fungsi jendela berbasis waktu kejadian?

Bagaimana cara menyelesaikan masalah kehilangan data?

Pengurangan volume data sering kali disebabkan oleh operasi penyaringan seperti klausa WHERE, JOIN, atau operasi berbasis jendela. Untuk menyelidiki kehilangan data yang tidak normal, lakukan hal berikut:

• Periksa kebijakan cache untuk tabel dimensi.

  Konfigurasikan kebijakan cache yang sesuai untuk tabel dimensi untuk mencegah kegagalan join lookup dan kehilangan data. Untuk informasi lebih lanjut, lihat opsi spesifik-dimensi terkait cache dari dokumen konektor terkait (seperti bagian "Opsi spesifik-dimensi (terkait cache)" dari topik Konektor ApsaraDB for HBase).

• Verifikasi fungsi digunakan dengan benar.

  Penggunaan fungsi yang salah seperti to_timestamp_tz dan date_format dapat menyebabkan kesalahan konversi data dan kehilangan data.

  Verifikasi penggunaan fungsi melalui analisis log dengan menggunakan tabel sink cetak atau Log4j. Untuk informasi lebih lanjut, lihat Print connector dan Konfigurasi parameter untuk mengekspor log dari suatu deployment.

• Periksa apakah ada data tidak berurutan.

  Peristiwa terlambat di luar kerangka waktu jendela saat ini dibuang. Dalam gambar berikut, peristiwa dengan timestamp 11s memasuki jendela 15-20s. Karena watermark-nya adalah 11, dianggap terlambat dan dibuang.

  Data biasanya hilang selama satu jendela waktu. Anda dapat menggunakan tabel sink cetak atau Log4j untuk mendeteksi data yang tidak berurutan. Untuk informasi lebih lanjut, lihat Print connector dan Konfigurasikan parameter untuk mengekspor log dari suatu deployment.

  Untuk menangani data tidak berurutan secara akurat, konfigurasikan strategi pembuatan watermark yang sesuai (seperti Watermark = Waktu kejadian - 5s). Selain itu, kami merekomendasikan menyelaraskan jendela ke interval hari, jam, atau menit yang tepat. Praktik ini, bersama dengan peningkatan periode tenggang, membantu mencegah data sangat tidak berurutan agar tidak hilang.

Mengapa saya mendapatkan hasil yang tidak akurat saat menggunakan ROW_NUMBER untuk menghapus duplikat data yang diambil dari Hologres dalam mode CDC?

• Hasil tidak akurat

  

• Penyebab

  Downstream menggunakan operator retraksi (seperti ROW_NUMBER OVER WINDOW untuk deduplikasi), tetapi data tidak diambil dari Hologres dalam mode upsert.

• Solusi

  Tambahkan 'upsertSource' = 'true' dalam klausa WITH dari pernyataan DDL tabel sumber untuk deduplikasi data.

  

Bagaimana cara menyelesaikan masalah hasil yang tidak akurat?

1. Atur ulang level log.

   Ubah level log menjadi INFO.

2. Aktifkan profil operator.

   Anda dapat melihat hasil perantara tanpa memodifikasi logika program.

3. Analisis log runtime.

   1. Klik nama deployment target.

   2. Di halaman detail deployment, klik tab Status.

   3. Di grafik DAG, salin nama operator.

   4. Dalam daftar log deployment, untuk kolom Log Name, klik inspect-taskmanager_0.out, dan cari nama operator.

   

4. Lakukan optimisasi dan verifikasi.

   Setelah menemukan penyebab dalam log, revisi logika operator yang bermasalah, mulai ulang pekerjaan, dan verifikasi akurasi data.

Bagaimana cara memperbaiki kesalahan "tidak mendukung konsumsi perubahan pembaruan dan penghapusan yang dihasilkan oleh node TableSourceScan"?

• Pesan kesalahan

  Table sink 'vvp.default.***' doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[vvp, default, ***]], fields=[id,b, content])
      at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.wrapExecutor(DelegateOperationExecutor.java:286)
      at org.apache.flink.table.sqlserver.execution.DelegateOperationExecutor.validate(DelegateOperationExecutor.java:211)
      at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.validate(FlinkSqlServiceImpl.java:741)
      at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:2522)
      at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
      at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)
      at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:820)
      at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
      at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
      at java.lang.Thread.run(Thread.java:834)

• Penyebab

  Mode penulisan tabel sink adalah append-only, sehingga tidak dapat mengonsumsi pembaruan data.

• Solusi

  Buat tabel sink yang mendukung peristiwa pembaruan menggunakan konektor seperti Upsert Kafka connector.

Bagaimana cara memperbaiki penimpaan atau penghapusan data yang tidak terduga saat menggunakan konektor Lindorm?

SET 'table.exec.sink.upsert-materialize' = 'NONE';