Hologres:Flink を使用した Hologres へのバッチインポート

背景情報

Hologres は Apache Flink と統合され、リアルタイムのデータストリーミング機能を提供します。一方で、既存データの読み込み、オフラインデータの処理、ログの集計など、時間的制約のないユースケースでは、バッチインポートが推奨されます。バッチインポートは大量のデータを一度に Hologres に書き込むため、より効率的でリソースを節約できます。ビジネス要件および利用可能なリソースに基づき、リアルタイムインポートとバッチインポートのいずれかを選択してください。リアルタイムインポートの詳細については、「Realtime Compute for Apache Flink」をご参照ください。

前提条件

• Hologres インスタンスを購入済みです。詳細については、「Hologres インスタンスを購入する」をご参照ください。

• Apache Flink クラスター（バージョン 1.15 以降）をデプロイ済みである必要があります。詳細については、以下のトピックをご参照ください。

  • Apache Flink: Flink のデプロイ。

  • Realtime Compute for Apache Flink：Realtime Compute for Apache Flink の有効化。

Realtime Compute for Apache Flink を使用したバッチインポート

1. Flink からインポートされたデータを格納するための Hologres 結果テーブルを作成します。手順については、「HoloWeb への接続とクエリの実行」をご参照ください。本トピックでは、test_sink_customer テーブルを例として使用します。

   -- Hologres 結果テーブルを作成します。
   CREATE TABLE test_sink_customer
   (
     c_custkey     BIGINT,
     c_name        TEXT,
     c_address     TEXT,
     c_nationkey   INT,
     c_phone       TEXT,
     c_acctbal     NUMERIC(15,2),
     c_mktsegment  TEXT,
     c_comment     TEXT,
     "date"        DATE
   ) WITH (
     distribution_key="c_custkey,date", 
     orientation="column"
   );

   説明

   Flink ソーステーブルのフィールド名およびデータの型は、Hologres 結果テーブルのものと一致している必要があります。

2. Realtime Compute for Apache Flink コンソールにログインします。Deployments ページで、Create Deployment をクリックします。デプロイメントパラメーターを設定し、デプロイメント をクリックします。パラメーターの詳細については、「JAR タスクのデプロイ」をご参照ください。

   以下の表は、主なパラメーターについて説明しています。

   パラメーター	説明
   Deployment Type	JAR を選択します。
   デプロイモード	ストリームモードまたはバッチモードを選択できます。本トピックでは、バッチモードを例として使用します。
   データベースエンジンバージョン	エンジンバージョンの詳細については、「エンジンバージョン」および「ライフサイクルポリシー」をご参照ください。本トピックでは、バージョン vvr-8.0.7-flink-1.17 を例として使用します。
   JAR URI	オープンソースの Flink コネクタをアップロードします：hologres-connector-flink-repartition.jar。 説明 オープンソースの Flink コネクタを使用して、Hologres にデータをバッチインポートできます。Flink コネクタのソースコードについては、公式 Hologres GitHub リポジトリをご参照ください。
   Entry Point Class	プログラムのエントリポイントクラスです。Flink コネクタのメインクラスは com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample です。
   Entry Point Main Arguments	repartition.sql ファイルへのパスを指定します。Realtime Compute for Apache Flink のランタイム環境では、追加の依存ファイルは /flink/usrlib/ に格納されるため、完全な引数は --sqlFilePath="/flink/usrlib/repartition.sql" となります。
   Additional Dependencies	repartition.sql ファイルをアップロードします。このファイルは、データソースの定義、結果テーブルの宣言、Hologres への接続設定を行う Flink SQL スクリプトです。以下のコードは、repartition.sql ファイルの例です。 -- ソーステーブルの DDL。この例では、テストデータを生成するために Flink DataGen コネクタを使用します。 CREATE TEMPORARY TABLE source_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ) WITH ( 'connector' = 'datagen' ,'rows-per-second' = '10000' ,'number-of-rows' = '1000000' ); -- ソーステーブルの DQL。クエリ結果は、シンク DDL で定義された結果テーブルのスキーマ（フィールド数およびデータの型を含む）と一致している必要があります。 SELECT *, cast('2024-04-21' as DATE) FROM source_table; -- シンクテーブルの DDL。結果テーブルを宣言し、Hologres への接続を設定します。 CREATE TABLE sink_table ( c_custkey BIGINT ,c_name STRING ,c_address STRING ,c_nationkey INTEGER ,c_phone STRING ,c_acctbal NUMERIC(15, 2) ,c_mktsegment STRING ,c_comment STRING ,`date` DATE ) WITH ( 'connector' = 'hologres' ,'dbname' = 'doc_****' ,'tablename' = 'test_sink_customer' ,'username' = 'yourAccessKeyId' ,'password' = 'yourAccessKeySecret' ,'endpoint' = 'hgpostcn-cn-7pp2e1k7****-cn-hangzhou.hologres.aliyuncs.com:80' ,'jdbccopywritemode' = 'true' ,'bulkload' = 'true' ,'target-shards.enabled'='true' ); 説明 repartition.sql ファイル内の Hologres 接続パラメーターの詳細については、「Hologres Flink コネクタパラメーター」をご参照ください。

3. デプロイメント名をクリックし、Deployment Details ページに移動します。リソース設定 セクションで、Parallelism を変更します。

   説明

   並列度は、Hologres 結果テーブルの Shard Count に設定することを推奨します。

4. Hologres 結果テーブルをクエリします。

   Flink タスクが送信された後、Hologres に書き込まれたデータをクエリできます。サンプル文：

   SELECT * FROM test_sink_customer;

Apache Flink を使用したバッチインポート

1. Flink からインポートされたデータを格納するための Hologres 結果テーブルを作成します。手順については、「HoloWeb への接続とクエリの実行」をご参照ください。本トピックでは、test_sink_customer テーブルを例として使用します。

   -- Hologres 結果テーブルを作成します。
   CREATE TABLE test_sink_customer
   (
     c_custkey     BIGINT,
     c_name        TEXT,
     c_address     TEXT,
     c_nationkey   INT,
     c_phone       TEXT,
     c_acctbal     NUMERIC(15,2),
     c_mktsegment  TEXT,
     c_comment     TEXT,
     "date"        DATE
   ) WITH (
     distribution_key="c_custkey,date", 
   
     orientation="column"
   );

   説明

   シャード数はデータ量に基づいて設定できます。シャードの詳細については、「テーブルグループおよびシャード数の管理」をご参照ください。

2. repartition.sql ファイルを作成し、Flink クラスター環境の任意のディレクトリにアップロードします。本トピックでは、/flink-1.15.4/src/repartition.sql を例として使用します。以下のコードは、repartition.sql ファイルの例です。

   説明

   このファイルは、データソースの定義、結果テーブルの宣言、Hologres への接続設定を行う Flink SQL スクリプトです。

   -- ソーステーブルの DDL。この例では、テストデータを生成するために Flink DataGen コネクタを使用します。
   CREATE TEMPORARY TABLE source_table
   (
     c_custkey     BIGINT
     ,c_name       STRING
     ,c_address    STRING
     ,c_nationkey  INTEGER
     ,c_phone      STRING
     ,c_acctbal    NUMERIC(15, 2)
     ,c_mktsegment STRING
     ,c_comment    STRING
   )
   WITH (
     'connector' = 'datagen'
     ,'rows-per-second' = '10000'
     ,'number-of-rows' = '1000000'
   );
   
   -- ソーステーブルの DQL。クエリ結果は、シンク DDL で定義された結果テーブルのスキーマ（フィールド数およびデータの型を含む）と一致している必要があります。
   SELECT *, cast('2024-04-21' as DATE) FROM source_table;
   
   -- シンクテーブルの DDL。結果テーブルを宣言し、Hologres への接続を設定します。
   CREATE TABLE sink_table
   (
     c_custkey     BIGINT
     ,c_name       STRING
     ,c_address    STRING
     ,c_nationkey  INTEGER
     ,c_phone      STRING
     ,c_acctbal    NUMERIC(15, 2)
     ,c_mktsegment STRING
     ,c_comment    STRING
     ,`date`       DATE
   )
   WITH (
     'connector' = 'hologres'
     ,'dbname' = 'doc_****'
     ,'tablename' = 'test_sink_customer'
     ,'username' = 'yourAccessKeyId'
     ,'password' = 'yourAccessKeySecret'
     ,'endpoint' = 'hgpostcn-cn-7pp2e1k7****-cn-hangzhou.hologres.aliyuncs.com:80'
     ,'jdbccopywritemode' = 'true'
     ,'bulkload' = 'true'
     ,'target-shards.enabled'='true'
   );

   以下の表は、主なパラメーターについて説明しています。

   パラメーター	必須	説明
   connector	はい	コネクタのタイプです。値は hologres である必要があります。
   dbname	はい	Hologres データベースの名前です。
   tablename	はい	データを受信する Hologres テーブルの名前です。
   username	はい	ご利用の Alibaba Cloud アカウントの AccessKey ID です。 AccessKey ID は、AccessKey Pair ページから取得できます。
   password	はい	AccessKey ID に対応する AccessKey Secret です。
   endpoint	はい	Hologres インスタンスの VPC エンドポイントです。Hologres コンソールのインスタンス詳細ページで、インスタンスの設定 セクションからエンドポイントを取得します。 説明 エンドポイントには、ip:port 形式でポート番号を含める必要があります。同一リージョン内での接続には VPC エンドポイントを、クロスリージョン接続にはパブリックエンドポイントを使用します。
   jdbccopywritemode	いいえ	データ書き込み方法です。有効な値は次のとおりです。 false（デフォルト）：INSERT メソッドを使用します。 true：COPY メソッドを使用します。COPY メソッドには、ストリーミング COPY（Fixed Copy）とバッチ COPY があります。デフォルトでは、ストリーミング COPY が使用されます。 説明 INSERT メソッドと比較して、ストリーミング COPY はストリーミングモデルを使用することで、より高いスループット、低いデータ遅延、およびクライアントのメモリ消費の削減（データがバッチ処理されないため）を実現します。ただし、データの取り消しはサポートされていません。
   bulkload	いいえ	バッチ COPY メソッドを使用するかどうかを指定します。有効な値は次のとおりです。 true：バッチ COPY を使用します。この設定は、jdbccopywritemode も true に設定されている場合にのみ有効です。それ以外の場合は、ストリーミング COPY が使用されます。 説明 ストリーミング COPY と比較して、バッチ COPY はより効率的で、Hologres リソースをより効果的に使用し、優れた書き込みパフォーマンスを実現します。ビジネス要件に基づき、適切な書き込み方法を選択してください。 プライマリキーを持つテーブルにバッチ COPY を使用してデータを書き込む場合、テーブルロックが発生する可能性があります。target-shards.enabled パラメーターを true に設定すると、ロックの粒度をテーブルレベルからシャードレベルに縮小できます。これにより、複数のバッチインポートタスクを同時に実行でき、テーブルロックの競合を軽減します。ストリーミング COPY と比較して、プライマリキーを持つテーブルへの書き込み時の Hologres インスタンスへの負荷が大幅に軽減されます。テストでは、負荷が約 66.7% 軽減されました。 バッチ COPY を使用する場合、送信先テーブルにプライマリキーが存在する場合は、書き込み操作前にテーブルが空である必要があります。そうでない場合、プライマリキーに基づくデータ重複排除により、書き込みプロセスが遅くなります。 false（デフォルト）：バッチ COPY を使用しません。
   target-shards.enabled	いいえ	ターゲットシャードへのバッチ書き込みを有効にするかどうかを指定します。有効な値は次のとおりです。 true：ターゲットシャードへのバッチ書き込みを有効にします。ソースデータがシャード単位で再パーティション化されている場合、ロックの粒度をシャードレベルに縮小できます。 false（デフォルト）：この機能を無効にします。

   説明

   repartition.sql ファイル内の Hologres 接続パラメーターの詳細については、「Hologres Flink コネクタパラメーター」をご参照ください。

3. Flink クラスター環境で、オープンソースの Flink コネクタ hologres-connector-flink-repartition.jar を任意のディレクトリにアップロードします。本トピックでは、ルートディレクトリを例として使用します。

   説明

   オープンソースの Flink コネクタを使用して、Hologres にデータをバッチインポートできます。Flink コネクタのソースコードについては、公式 Hologres GitHub リポジトリをご参照ください。

4. Flink タスクを送信します。サンプルコマンド：

   ./bin/flink run -Dexecution.runtime-mode=BATCH -p 3 -c com.alibaba.ververica.connectors.hologres.example.FlinkToHoloRePartitionExample hologres-connector-flink-repartition.jar --sqlFilePath="/flink-1.15.4/src/repartition.sql"

   上記コマンドのパラメーター：

   • Dexecution.runtime-mode: Flink ジョブの実行モード。詳細については、「実行モード」をご参照ください。

   • p：タスクの並列度です。この値は、結果テーブルの Shard Count またはその約数に設定することを推奨します。

   • c：hologres-connector-flink-repartition.jar ファイル内のメインクラスの完全修飾名です。

   • sqlFilePath：repartition.sql ファイルへのパスです。

5. Hologres 結果テーブルをクエリします。

   Flink タスクが送信された後、Hologres に書き込まれたデータをクエリできます。サンプル文：

   SELECT * FROM test_sink_customer;