Realtime Compute for Apache Flink：Konektor Hologres untuk ingesti data (YAML)

Pemetaan tipe data

Gunakan parameter sink.type-normalize-strategy untuk menentukan cara tipe data upstream dipetakan ke tipe data Hologres.

LENIENT (default)

Dalam mode LENIENT, perubahan skema ditangani sebagai berikut:

• Menambahkan kolom nullable: kolom ditambahkan ke akhir skema tabel sink dan data disinkronkan ke kolom tersebut.

• Menghapus kolom nullable: kolom diisi dengan null alih-alih dihapus dari tabel sink.

• Menambahkan kolom non-nullable: kolom ditambahkan ke akhir skema tabel sink dan data disinkronkan ke kolom tersebut. Kolom baru secara default dianggap nullable, dan data sebelum penambahan kolom diisi dengan null.

• Mengganti nama kolom: diperlakukan sebagai operasi tambah + hapus. Kolom yang diganti namanya ditambahkan ke akhir tabel sink, sedangkan kolom asli diisi dengan null. Misalnya, jika col_a diubah menjadi col_b, maka col_b ditambahkan dan col_a diisi dengan null.

• Mengubah tipe kolom: tidak didukung. Hologres tidak mengizinkan perubahan tipe kolom; gunakan sink.type-normalize-strategy sebagai gantinya.

• Perubahan skema berikut tidak didukung:

  • Perubahan pada constraint seperti kunci primary atau indeks.

  • Penghapusan kolom non-nullable.

  • Mengubah dari NOT NULL menjadi nullable.

EVOLVE

Dalam mode EVOLVE, perubahan skema ditangani sebagai berikut:

• Menambahkan kolom nullable: didukung.

• Menghapus kolom nullable: tidak didukung.

• Kolom non-null ditambahkan ke tabel sink sebagai kolom nullable.

• Mengganti nama kolom: didukung; nama kolom di tabel sink diperbarui sesuai.

• Mengubah tipe kolom: tidak didukung. Hologres tidak mengizinkan perubahan tipe kolom; gunakan sink.type-normalize-strategy sebagai gantinya.

• Perubahan skema berikut tidak didukung:

  • Perubahan pada constraint seperti kunci primary atau indeks.

  • Penghapusan kolom non-nullable.

  • Mengubah dari NOT NULL menjadi nullable.

Pemetaan Tipe Luas

Gunakan parameter sink.type-normalize-strategy untuk memperluas pemetaan tipe data.

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

Tulis ke tabel partisi

Konversi field timestamp upstream create_time ke tipe date dan gunakan sebagai kunci partisi 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}
  # Membuat tabel partisi yang hilang secara otomatis.
  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

Aktifkan mode EVOLVE

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}
  # Membuat tabel partisi yang hilang secara otomatis.
  createparttable: true

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

Sinkronisasi satu tabel

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 semua tabel dalam database

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

Gabungkan tabel 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:
  # Gabungkan semua tabel sharded di MySQL test_db menjadi satu tabel Hologres: test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Sinkronisasi ke skema tertentu

Skema di Hologres berkorespondensi 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 MySQL test_db ke skema Hologres test_db2, tanpa mengubah nama tabel.
  - 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 dibuat secara real time selama pekerjaan berjalan, atur scan.binlog.newly-added-table.enabled ke true.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # Tangkap data dari tabel baru yang dibuat selama 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

Restart Tabel Eksisting yang Baru Ditambahkan

Jika Anda ingin menambahkan sinkronisasi untuk tabel eksisting, atur scan.newly-added-table.enabled = 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, periksa parameter tables untuk tabel baru dan jalankan snapshot.
  # Memerlukan 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

Kecualikan tabel selama sinkronisasi seluruh database

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Tabel yang cocok dengan regex ini tidak akan disinkronkan.
  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