Pengumpulan data kotor - Realtime Compute for Apache Flink

Topik ini menjelaskan cara menggunakan pengumpul data kotor dalam pekerjaan ingesti data Flink CDC.

Ikhtisar

Dalam sinkronisasi data real-time, data dari sumber dapat gagal diurai karena masalah seperti format yang salah, error pengkodean, atau ketidakcocokan skema. Data yang tidak dapat diproses ini dikenal sebagai dirty data.

Mulai dari versi VVR 11.5, ingesti data mendukung pengumpulan data kotor untuk sumber data Kafka. Opsi konfigurasi konektor memungkinkan Anda mengonfigurasi pekerjaan agar mengabaikan error, mencatat detailnya, dan terus berjalan.

Ketika konektor menemui data yang tidak dapat diurai, sistem secara otomatis menangkap data mentah dan informasi exception, lalu menuliskannya ke pengumpul yang ditentukan. Hal ini memungkinkan Anda untuk:

Menoleransi sejumlah kecil data kotor agar seluruh pipeline tidak terganggu.
Mencatat konteks lengkap untuk troubleshooting dan analisis di kemudian hari.
Menetapkan ambang batas untuk mencegah jumlah error yang berlebihan.

Kasus penggunaan khas

Kasus penggunaan	Tujuan
Pipeline pengumpulan log (Untuk sumber data tidak terstruktur seperti log aplikasi)	Kualitas data tidak konsisten. Lewati beberapa catatan buruk agar proses utama tetap berjalan.
Sinkronisasi tabel bisnis inti (Untuk sistem kritis seperti pesanan atau perubahan akun)	Memerlukan konsistensi tinggi. Error memicu notifikasi segera untuk intervensi manual.
Fase eksplorasi dan analisis data	Memproses seluruh dataset dengan cepat untuk memahami struktur keseluruhan sebelum menangani isu data kotor.

Batasan dan pertimbangan

Sebelum menggunakan fitur ini, pahami batasan dan risiko potensialnya:

Konektor yang didukung: Fitur ini saat ini hanya tersedia untuk sumber data Kafka. Dukungan untuk sumber lain akan ditambahkan pada rilis mendatang.
Jenis pengumpul yang didukung: Saat ini hanya tipe logger yang didukung, yang menuliskan data kotor ke file log.

Catatan

Fitur ini cocok untuk debugging dan tahap produksi awal. Jika jumlah data kotor tetap besar, kami merekomendasikan menerapkan langkah tata kelola data pada sistem hulu.

Sintaks

Aktifkan pengumpul data kotor

Pengumpul data kotor didefinisikan dalam modul pipeline. Sintaksnya adalah sebagai berikut:

pipeline:
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

Parameter

Deskripsi

name

Nama pengumpul. Kami menyarankan menggunakan nama yang bermakna, seperti Kafka-DQ-Collector.

type

Jenis pengumpul. Nilai yang valid:

logger: Menuliskan data kotor ke file log.

Catatan

Jika Anda tidak menentukan opsi ini, sistem tidak akan mencatat data kotor, meskipun toleransi error diaktifkan.

Konfigurasikan toleransi error pada sumber data

Hanya mengonfigurasi pengumpul data kotor tidak cukup untuk melewatkan error penguraian. Anda harus menggunakan fitur ini bersamaan dengan kebijakan toleransi error Kafka. Untuk detailnya, lihat dokumentasi konektor Kafka. Contoh berikut menunjukkan konfigurasi khas:

source:
  type: kafka
  # Lewati 100 error penguraian pertama; jika jumlahnya melebihi 100, pekerjaan gagal.
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

Parameter

Default

Deskripsi

ingestion.ignore-errors

false

Menentukan apakah akan mengabaikan error penguraian.

Jika diatur ke true, pekerjaan melewatkan catatan yang gagal. Jika diatur ke false, pekerjaan langsung gagal.

ingestion.error-tolerance.max-count

-1 (tidak terbatas)

Jumlah maksimum catatan data kotor yang dapat ditoleransi.

Ketika ingestion.ignore-errors diatur ke true, jika jumlah catatan data kotor yang dikumpulkan melebihi nilai ini, pekerjaan memicu failover dan berhenti.

Pengumpul data kotor logger

Pengumpul data kotor logger menyimpan data kotor dalam file log terpisah. Untuk melihat log data kotor, ikuti langkah-langkah berikut:

Buka halaman O&M dan klik tab Job Logs.
Klik Running Logs, pilih sub-tab Running Task Managers, lalu pilih node TaskManager untuk operator terkait.
Klik Log List, lalu klik file log bernama yaml-dirty-data.out dalam daftar untuk melihat atau menyimpan catatan data kotor.

Setiap catatan data kotor mencakup metadata berikut:

Timestamp saat data kotor diproses
Operator dan indeks subtask yang menghasilkan catatan data kotor
Konten data mentah
Informasi exception yang menyebabkan kegagalan pemrosesan

Format catatan data kotor

Setiap catatan berisi metadata berikut:

text[2025-04-05 10:23:45] [Operator: SourceKafka -> Subtask: 2]
Raw Data: {"id": "abc", "ts": "invalid-timestamp"}
Exception: java.time.format.DateTimeParseException: Text 'invalid-timestamp' could not be parsed at index 0
---

Field	Deskripsi
Timestamp	Waktu saat data kotor ditangkap.
Operator & subtask	Operator spesifik dan indeks subtask paralelnya tempat error terjadi.
Raw data	Data mentah yang belum diurai (dalam format Base64 atau string).
Exception	Jenis exception dan ringkasan stack trace untuk kegagalan penguraian.

FAQ

Apakah data kotor memengaruhi checkpoint?

Tidak. Data kotor dicegat sebelum dimasukkan ke dalam pembaruan state, sehingga tidak menyebabkan checkpoint gagal.

Pengumpulan data kotor vs. side output

Pengumpul data kotor: Menangani data yang tidak dapat dideserialisasi atau diurai.
Side output: Menangani data yang dapat diurai tetapi tidak memenuhi persyaratan logika bisnis.