Realtime Compute for Apache Flink:データインジェスト用 Hologres YAML コネクタ

既存のカタログの再利用

VVR 11.5 以降では、データ管理ページで作成された組み込みの Hologres カタログを Flink CDC データインジェスト ジョブで直接参照できます。これにより、接続プロパティを手動で指定する手間が省けます。

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

データインジェスト ジョブは、次の 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 から NULLABLE に変更。

EVOLVE

EVOLVE モードでは、スキーマの変更は次のように処理されます。

• NULL 許容列の追加：サポートされています。

• NULL 許容列の削除：サポートされていません。

• NOT NULL 列の追加：新しい NULL 許容列がシンクテーブルに追加されます。

• 列の名前変更：サポートされています。元の列はシンクテーブルで名前が変更されます。

• 列のデータ型の変更：サポートされていません。Hologres は列のデータ型の変更をサポートしていないため、sink.type-normalize-strategy パラメーターを使用する必要があります。

• 以下のスキーマ変更はサポートされていません。

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

  • NOT NULL 列の削除。

  • 列を NOT NULL から NULLABLE に変更。

型拡張

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 to Hologres Pipeline

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

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 to Hologres Pipeline

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 to Hologres Pipeline
  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 to Hologres Pipeline

完全なデータベース同期

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 to Hologres Pipeline

シャードテーブルのマージ

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 データベース内のすべてのシャードテーブルを、test_db.user という名前の単一の Hologres テーブルにマージします。
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

指定スキーマへの同期

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

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
  
route:
  # 元のテーブル名を維持したまま、MySQL の test_db データベースからすべてのテーブルを Hologres の test_db2 スキーマに同期します。
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

再起動なしでの新しいテーブルの同期

ジョブの実行中に新しく追加されたテーブルをリアルタイムで同期するには、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 to Hologres Pipeline

再起動時に既存のテーブルを追加

同期に既存のテーブルを含めるには、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 to Hologres Pipeline

テーブルの除外

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 to Hologres Pipeline