Realtime Compute for Apache Flink:YAML デプロイメントでの Hologres コネクタを使用したデータインジェスト (パブリックプレビュー)

データ型のマッピング

sink.type-normalize-strategy 設定項目を使用して、アップストリームのデータ型を Hologres のデータ型に変換するポリシーを設定できます。

LENIENT（デフォルト値）

schema.change.behavior を LENIENT に設定すると、テーブルスキーマの変更は、次の操作の説明に基づいて同期されます。

• NULL 値が許可された列を追加する：ステートメントは、関連する列をシンクテーブルのスキーマの最後に自動的に追加し、追加された列にデータを同期します。

• NULL 値が許可された列を削除する：ステートメントは、テーブルから列を削除する代わりに、シンクテーブルの NULL 値が許可された列に NULL 値を自動的に設定します。

• NULL 値が許可されていない列を追加する：ステートメントは、関連する列をシンクテーブルのスキーマの最後に自動的に追加し、新しい列のデータを同期します。新しい列は自動的に NULL 値が許可された列に設定され、列が追加される前のデータは自動的に NULL 値に設定されます。

• 列の名前を変更する：列の名前を変更する操作には、列の追加と列の削除が含まれます。ソーステーブルで列の名前が変更されると、新しい名前を使用する列がシンクテーブルの最後に追加され、元の名前を使用する列には NULL 値が設定されます。たとえば、ソーステーブルの col_a 列の名前が col_b に変更された場合、col_b 列がシンクテーブルの最後に追加され、col_a 列には自動的に NULL 値が設定されます。

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

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

  • 主キーやインデックスなどの制約の変更

  • NULL 値が許可されていない列の削除

  • NULL 値が許可されていない列から NULL 値が許可された列への変更

EVOLVE

schema.change.behavior を EVOLVE に設定すると、テーブルスキーマの変更は、次の操作の説明に基づいて同期されます。

• NULL 値が許可された列を追加する：サポートされています。

• NULL 値が許可された列を削除する：サポートされていません。

• NULL 値が許可されていない列を追加する：NULL 値が許可された列がシンクテーブルに追加されます。

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

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

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

  • 主キーやインデックスなどの制約の変更

  • NULL 値が許可されていない列の削除

  • 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}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

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

timestampデータ型のアップストリームフィールド create_time を dateデータ型のフィールドに変換し、新しいフィールドを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

有効化モード

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}
  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}
  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}
  sink.type-normalize-strategy: BROADEN
  
route:
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

特定のテーブルのスキーマの同期

Hologresテーブルのスキーマは、MySQLテーブルのデータベースに対応しています。 特定の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}
  sink.type-normalize-strategy: BROADEN
  
route:
  - source-table: test_db.user\.*
    sink-table: test_db2.user\.*r

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}
  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.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}
  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}
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline