Realtime Compute for Apache Flink：Distribusi data real-time menggunakan Flink CDC

Sinkronkan data binary logging MySQL ke Kafka

Dalam beberapa skenario, Anda mungkin perlu menyimpan data binary logging mentah untuk tugas seperti audit atau replay data. Pekerjaan YAML ingesti data mendukung sinkronisasi data binary logging MySQL mentah ke Kafka, memungkinkan pembacaan data secara terdistribusi dan membantu mengatasi masalah hot spot data.

Misalnya, database kafka_test berisi dua tabel, customers dan products. Pekerjaan berikut menyinkronkan data dari kedua tabel tersebut ke dua topik dengan nama yang sama: customers dan products.

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: kafka_test.\.*
  server-id: 8601-8604
  # (Opsional) Sinkronkan data lengkap dan inkremental dari tabel yang baru ditambahkan.
  scan.newly-added-table.enabled: true
  # (Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  # (Opsional) Utamakan mendistribusikan chunk tak terbatas untuk menghindari potensi kesalahan OutOfMemory pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Opsional) Aktifkan filter penguraian untuk mempercepat pembacaan.
  scan.only.deserialize.captured.tables.changelog.enabled: true
  # Kirim waktu perubahan data sebagai metadata.
  metadata-column.include-list: op_ts

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  # Alibaba Cloud Kafka tidak mendukung penulisan idempoten atau transaksional. Nonaktifkan fitur idempotensi.
  properties.enable.idempotence: false
  # (Opsional) Tetapkan pemetaan antara tabel hulu dan topik Kafka.
  sink.tableId-to-topic.mapping: kafka_test.customers:customers;kafka_test.products:products

Kode berikut menunjukkan format isi pesan Kafka yang dihasilkan oleh pernyataan UPDATE pada tabel customers:

// debezium-json
{
  "before": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "2222",
    "age": 12
  },
  "after": {
    "id": 4,
    "name": "John",
    "address": "New York",
    "phone_number": "1234",
    "age": 12
  },
  "op": "u",
  "source": {
    "db": "kafka_test",
    "table": "customers",
    "ts_ms": 1728528674000
  }
}

// canal-json
{
  "old": [
    {
      "id": 4,
      "name": "John",
      "address": "New York",
      "phone_number": "2222",
      "age": 12
    }
  ],
  "data": [
    {
      "id": 4,
      "name": "John",
      "address": "New York",
      "phone_number": "1234",
      "age": 12
    }
  ],
  "type": "UPDATE",
  "database": "kafka_test",
  "table": "customers",
  "pkNames": [
    "id"
  ],
  "ts": 1728528674000,
  "es": 0
}

Ingesti real-time dari Kafka ke DLF

Menyinkronkan data binary logging MySQL ke Kafka memberikan solusi untuk menyimpan data MySQL di Kafka. Selanjutnya, Anda dapat mengonfigurasi pekerjaan YAML ingesti data untuk menyinkronkan data tersebut ke penyimpanan DLF.

Misalnya, topik Kafka bernama inventory berisi data dari dua tabel, customers dan products, dalam format debezium-json. Pekerjaan berikut menyinkronkan data dari kedua tabel tersebut ke tabel yang sesuai di DLF:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Opsional) Nama pengguna untuk mengirimkan pekerjaan. Tetapkan nama pengguna unik untuk setiap pekerjaan guna menghindari konflik.
  commit.user: your_job_name
  # (Opsional) Aktifkan deletion vectors untuk meningkatkan kinerja baca.
  table.properties.deletion-vectors.enabled: true

# debezium-json tidak menyertakan informasi kunci primer. Anda harus menambahkan kunci primer ke tabel.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id