Hologres:Realtime Compute for Apache Flink または Blink を使用して Hologres バイナリログデータをリアルタイムで消費する

Hologres バイナリログを消費する前に、以下の項目に注意してください。

• Hologres V0.9 以降では、バイナリログを消費できます。 Hologres V1.3.21 以降では、エンジンのホワイトリストを設定できます。 V1.3.21 より前のバージョンのインスタンスにエンジンのホワイトリストを設定すると、インスタンスのバイナリログを消費できません。 Hologres インスタンスのバージョンが V1.3.21 より前の場合は、技術サポートのために Hologres DingTalk グループに参加できます。 詳細については、Hologres のオンラインサポートを受けるをご参照ください。

• Hologres では、テーブルレベルでバイナリログを消費できます。 行指向テーブルと列指向テーブルの両方がサポートされています。 Hologres インスタンスのバージョンが V1.1 以降の場合は、行と列のハイブリッドストレージモードを使用するテーブルのバイナリログも消費できます。 バイナリロギング機能を有効にすると、理論的には、列指向テーブルのオーバーヘッドは行指向テーブルよりも大きくなります。 そのため、データが頻繁に更新される場合は、行指向テーブルに対してバイナリロギング機能を有効にすることをお勧めします。

• バイナリロギング機能のサポートと、この機能を有効化および設定する方法の詳細については、Hologres バイナリログをサブスクライブするをご参照ください。

• Alibaba Cloud Realtime Compute for Apache Flink のみで Hologres バイナリログを消費できます。 Realtime Compute for Apache Flink を使用して HoloHub モードで Hologres バイナリログを消費する場合、単純なデータ型のみがサポートされます。 Ververica Runtime（VVR）6.0.3 以降を使用する Realtime Compute for Apache Flink では、Java Database Connectivity（JDBC）モードで Hologres バイナリログを消費できます。 HoloHub モードと比較して、JDBC モードはより多くのデータ型をサポートしています。 データ型のマッピングの詳細については、データ型の「Realtime Compute for Apache Flink または Blink と Hologres 間のデータ型のマッピング」セクションをご参照ください。 JDBC モードと HoloHub モードに必要な権限の詳細については、このトピックの権限をご参照ください。

• 親パーティションテーブルのバイナリログは消費できません。

• Hologres V2.0 および V2.0 以降のマイナーバージョンでは、特定の条件が満たされた場合に HoloHub モードがサポートされます。 Hologres V2.1 以降では、HoloHub モードはサポートされなくなりました。 JDBC モードのみがサポートされています。 Hologres インスタンスをアップグレードする前に、HoloHub モードを使用する Realtime Compute for Apache Flink デプロイメントを確認し、Realtime Compute for Apache Flink デプロイメントの VVR バージョンをアップグレードしてください。 詳細については、このトピックのHoloHub モードから JDBC モードへの変更をご参照ください。

このモードでは、ソーステーブルで消費されるバイナリログは、Realtime Compute for Apache Flink の通常のデータとしてダウンストリームノードに転送されます。 すべてのデータの変更タイプは INSERT です。 hg_binlog_event_type フィールドで指定されたタイプのデータを、ビジネス要件に基づいて処理できます。 Hologres テーブルに対してバイナリロギング機能が有効になったら、次の DDL ステートメントを実行して、Realtime Compute for Apache Flink で非変更データキャプチャ（CDC）モードでバイナリログを消費するソーステーブルを作成できます。

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>', /* データベース名 */
  'tablename'='<yourTablename>', /* テーブル名 */
  'username'='<yourAccessID>', /* AccessKey ID */
  'password'='<yourAccessSecret>', /* AccessKey Secret */
  'endpoint'='<yourEndpoint>', /* エンドポイント */
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

• binlog というプレフィックスが付いた 3 つのフィールドはシステムフィールドです。 これらのフィールドの名前またはデータ型を変更することはできません。

• 他のフィールドはユーザーフィールドに対応し、すべて小文字にする必要があります。

このモードでは、ソーステーブルで消費されるバイナリログデータの各行に、hg_binlog_event_type フィールドで指定されたタイプに基づいて、INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER などの正確な Flink RowKind タイプが自動的に割り当てられます。 これにより、バイナリログをソーステーブルにミラーリングできます。 これは、MySQL および PostgreSQL の CDC 機能に似ています。

Hologres テーブルに対してバイナリロギング機能が有効になったら、次の DDL ステートメントを実行して、Realtime Compute for Apache Flink で CDC ソーステーブルのバイナリログをリアルタイムで消費できます。

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>', // Hologres データベースの名前。
  'tablename'='<yourTablename>',// Hologres テーブルの名前。
  'username'='<yourAccessID>',// 現在のアカウントの AccessKey ID。
  'password'='<yourAccessSecret>',// 現在のアカウントの AccessKey シークレット。
  'endpoint'='<yourEndpoint>',// Hologres インスタンスの VPC エンドポイント。
  'binlog' = 'true',
  'cdcMode' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);