Realtime Compute for Apache Flink:Hologres コネクタによるデータインジェスト (YAML)

データ型のマッピング

sink.type-normalize-strategy パラメーターを用いて、上流のデータ型を Hologres のデータ型にマップする方法を定義します。

LENIENT（デフォルト）

LENIENT モードでは、スキーマ変更は以下のとおりに処理されます：

• NULL 許容カラムの追加：結果テーブルのスキーマ末尾にカラムを追加し、データを同期します。

• NULL 許容カラムの削除：カラムを結果テーブルから削除せず、代わりに null で埋めます。

• NOT NULL カラムの追加：結果テーブルのスキーマ末尾にカラムを追加し、データを同期します。新しいカラムはデフォルトで NULL 許容となります。カラムが追加される前のデータは、デフォルトで null になります。

• カラム名の変更：追加＋削除として扱われます。結果テーブルのスキーマ末尾に名前変更されたカラムを追加し、元のカラムを null で埋めます。たとえば、col_a が col_b に変更された場合、col_b が追加され、col_a は null で埋められます。

• カラム型の変更：非対応です。Hologres ではカラム型の変更は許可されていません。代わりに sink.type-normalize-strategy を使用してください。

• 以下のスキーマ変更は非対応です：

  • プライマリキーまたはインデックスなどの制約の変更。

  • NOT NULL カラムの削除。

  • NOT NULL から NULL 許容への変更。

EVOLVE

EVOLVE モードでは、スキーマ変更は以下のとおりに処理されます：

• NULL 許容カラムの追加：対応。

• NULL 許容カラムの削除：非対応。

• NOT NULL カラムは、結果テーブルに NULL 許容カラムとして追加されます。

• カラム名の変更：対応。結果テーブルのカラム名を変更します。

• カラム型の変更：非対応です。Hologres ではカラム型の変更は許可されていません。代わりに sink.type-normalize-strategy を使用してください。

• 以下のスキーマ変更は非対応です：

  • プライマリキーまたはインデックスなどの制約の変更。

  • NOT NULL カラムの削除。

  • NOT NULL から NULL 許容への変更。

広範な型マッピング

sink.type-normalize-strategy パラメーターを用いて、データ型のマッピングを広範なものにします。

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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン

パーティションテーブルへの書き込み

上流のタイムスタンプフィールド create_time を日付型に変換し、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}
  # 不足しているパーティションテーブルを自動的に作成します。
  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: パーティションキーの追加 

pipeline:
  name: MySQL から Hologres へのパイプライン

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}
  # 不足しているパーティションテーブルを自動的に作成します。
  createparttable: true

pipeline:
  name: MySQL から Hologres へのパイプライン
  schema.change.behavior: 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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン

データベース内のすべてのテーブルを同期

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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン

シャード化されたテーブルのマージ

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}
  # CDC データ型をより広範な Hologres の型にマップします。  
  sink.type-normalize-strategy: BROADEN
  
route:
  # MySQL test_db 内のすべてのシャード化されたテーブルを、1 つの Hologres テーブル test_db.user にマージします。
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL から Hologres へのパイプライン

特定のスキーマへの同期

Hologres のスキーマは、MySQL のデータベースに対応します。結果テーブルのスキーマを指定できます。

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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN
  
route:
  # MySQL test_db のすべてのテーブルを、テーブル名を変更せずに Hologres test_db2 スキーマに同期します。
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL から Hologres へのパイプライン

実行中のジョブで新規テーブルを同期

ジョブ実行中に新しく作成されたテーブルをリアルタイムで同期するには、scan.binlog.newly-added-table.enabled を true に設定します。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # ジョブ実行中に作成された新規テーブルからのデータをキャプチャします。
  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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン

新しく追加された既存テーブルの再起動

既存のテーブルの同期を追加する場合は、scan.newly-added-table.enabled = true を設定してジョブを再起動します。

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
  # 再起動時に、tables パラメーターで新規テーブルをチェックし、スナップショットを実行します。
  # 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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン

全データベース同期時のテーブル除外

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # この正規表現に一致するテーブルは同期されません。
  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}
  # CDC データ型をより広範な Hologres の型にマップします。
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL から Hologres へのパイプライン