Realtime Compute for Apache Flink:Konektor YAML Hologres untuk ingesti data

Gunakan Katalog yang Sudah Ada

Mulai dari VVR 11.5, Anda dapat langsung mereferensikan Katalog Hologres bawaan yang dibuat di halaman Data Management dalam pekerjaan ingesti data Flink CDC. Hal ini mengurangi kebutuhan untuk menentukan properti koneksi secara manual.

sink:
  type: hologres
  using.built-in-catalog: my_holo_catalog

Pekerjaan ingesti data dapat secara otomatis menggunakan kembali parameter Katalog Hologres berikut:

• endpoint

• username

• password

• dbname

Untuk mengganti parameter yang digunakan kembali secara otomatis ini, Anda dapat secara eksplisit menentukan parameter YAML yang sesuai, yang memiliki prioritas lebih tinggi.

Pemetaan tipe data

Anda dapat menggunakan parameter sink.type-normalize-strategy untuk mengatur strategi konversi data hulu ke tipe Hologres.

LENIENT (bawaan)

Dalam mode LENIENT, perubahan skema ditangani sebagai berikut:

• Menambahkan kolom nullable: Kolom yang sesuai secara otomatis ditambahkan di akhir tabel sink, dan datanya disinkronkan.

• Menghapus kolom nullable: Kolom tidak dihapus dari tabel sink. Sebaliknya, kolom tersebut secara otomatis diisi dengan nilai NULL.

• Menambahkan kolom non-nullable: Kolom nullable yang sesuai secara otomatis ditambahkan di akhir tabel sink, dan datanya disinkronkan. Untuk baris yang sudah ada, kolom baru ini secara otomatis diisi dengan NULL.

• Mengganti nama kolom: Operasi ini dianggap sebagai penghapusan kolom dan penambahan kolom baru. Kolom baru dengan nama yang ditentukan ditambahkan di akhir tabel sink, dan kolom dengan nama asli secara otomatis diisi dengan nilai NULL. Misalnya, jika col_a diganti namanya menjadi col_b, kolom col_b ditambahkan di akhir tabel sink, dan kolom col_a secara otomatis diisi dengan nilai NULL.

• Mengubah tipe data kolom: Tidak didukung. Karena Hologres tidak mendukung perubahan tipe data kolom, Anda harus menggunakan parameter sink.type-normalize-strategy.

• Perubahan skema berikut tidak didukung:

  • Perubahan pada constraint, seperti kunci primer atau indeks.

  • Menghapus kolom non-nullable.

  • Mengubah kolom dari NOT NULL menjadi NULLABLE.

EVOLVE

Dalam mode EVOLVE, perubahan skema ditangani sebagai berikut:

• Menambahkan kolom nullable: Didukung.

• Menghapus kolom nullable: Tidak didukung.

• Menambahkan kolom non-nullable: Kolom nullable baru ditambahkan ke tabel sink.

• Mengganti nama kolom: Didukung. Kolom asli diganti namanya di tabel sink.

• Mengubah tipe data kolom: Tidak didukung. Karena Hologres tidak mendukung perubahan tipe data kolom, Anda harus menggunakan parameter sink.type-normalize-strategy.

• Perubahan skema berikut tidak didukung:

  • Perubahan pada constraint, seperti kunci primer atau indeks.

  • Menghapus kolom non-nullable.

  • Mengubah kolom dari NOT NULL menjadi NULLABLE.

Pelebaran tipe

Gunakan parameter sink.type-normalize-strategy untuk mengonfigurasi pelebaran tipe.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Penulisan ke tabel partisi

Konversi bidang timestamp create_time ke tipe date dan gunakan sebagai kunci partisi untuk tabel Hologres.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Secara otomatis membuat tabel partisi jika belum ada.
  createparttable: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Hologres Pipeline

Mengaktifkan EVOLVE mode

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Secara otomatis membuat tabel partisi jika belum ada.
  createparttable: true

pipeline:
  name: MySQL to Hologres Pipeline
  schema.change.behavior: evolve

Sinkronisasi tabel tunggal

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Sinkronisasi database penuh

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Menggabungkan tabel terbagi (sharded)

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.  
  sink.type-normalize-strategy: BROADEN
  
route:
  # Semua tabel terbagi di database MySQL test_db digabung menjadi satu tabel Hologres bernama test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Sinkronisasi ke skema tertentu

Di Hologres, skema berpadanan dengan database di MySQL. Anda dapat menentukan skema untuk tabel sink.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN
  
route:
  # Sinkronkan semua tabel dari database MySQL test_db ke skema Hologres test_db2 sambil mempertahankan nama tabel asli.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

Sinkronisasi tabel baru tanpa restart

Untuk menyinkronkan tabel yang baru ditambahkan secara real-time saat pekerjaan sedang berjalan, atur scan.binlog.newly-added-table.enable = true.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # Secara otomatis menangkap tabel baru yang dibuat saat pekerjaan berjalan.  
  scan.binlog.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Menambahkan tabel yang sudah ada saat restart

Untuk menyertakan tabel yang sudah ada dalam sinkronisasi, atur scan.newly-added-table.enabled ke true dan restart pekerjaan.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # Saat restart, pekerjaan memindai tabel baru yang sesuai dengan parameter `tables` dan melakukan snapshot.
  # Catatan: Parameter ini harus digunakan dengan scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Mengecualikan tabel

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Mengecualikan tabel yang sesuai dengan ekspresi reguler ini.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Memetakan tipe data CDC ke tipe Hologres yang lebih luas.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline