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

既存のカタログの再利用

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

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

データインジェストジョブでは、次の Hologres カタログパラメーターを自動的に再利用できます：

• endpoint

• username

• password

• dbname

これらの自動再利用パラメーターを上書きするには、対応する YAML パラメーターを明示的に指定できます。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 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: add partition key 

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

再起動せずに新規テーブルを同期

ジョブ実行中に新しく追加されたテーブルをリアルタイムで同期するには、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
  # ジョブ実行中に新しく作成されたテーブルを自動的にキャプチャします。  
  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