Hologres:Hologres へのバッチ書き込み

環境のセットアップ

開始する前に、次の環境がセットアップ済みであることを確認してください。

• V2.2.25 以降の Hologres インスタンスをプロビジョニングして、データベースを作成します。

• Spark 3.3.0 以降を実行する EMR Spark クラスターを作成します。詳細については、「クラスターの作成」をご参照ください。

• Spark-Connector パッケージをダウンロードします。

  Spark から Hologres への読み取りと書き込みを有効にするには、Maven 中央リポジトリから hologres-connector-spark-3.x JAR ファイルをダウンロードします。

次の表は、性能テストに使用した環境を示しています。

テストデータの準備

1. ソースデータを準備します。

   1. EMR Spark クラスターのマスターノードにログオンします。詳細については、「ECS インスタンスへの接続」をご参照ください。

   2. TPC-H_Tools_v3.0.0.zip TPC-H ツールパッケージをダウンロードしてマスターノードの ECS インスタンスにコピーし、解凍してから TPC-H_Tools_v3.0.0/TPC-H_Tools_v3.0.0/dbgen ディレクトリに移動します。

   3. dbgen ディレクトリで次のコマンドを実行し、1 TB のテストデータセットとして customer.tbl ファイルを生成します。

      ./dbgen -s 1000 -T c

      次の表は、1 TB TPC-H データセット内の customer テーブルについて説明します。

      項目	説明
      フィールド数	8
      フィールドタイプ	INT、BIGINT、TEXT、DECIMAL
      行数	150,000,000
      シャード数	40

2. テストデータを Spark にインポートします。

   次のコマンドを実行して、customer.tbl ファイルを Spark クラスターにアップロードします。

   hadoop fs -put customer.tbl <spark_resource>

   コマンド内の spark_resource は、EMR Spark クラスターの作成時に設定した [クラスターのルートストレージディレクトリ] を指します。

3. 行-列ハイブリッドストレージ

   CREATE TABLE test_table_mixed (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column,row'
   );

   列指向ストレージ (プライマリキーあり)

   CREATE TABLE test_table_column (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column'
   );

   列指向ストレージ (プライマリキーなし)

   CREATE TABLE test_table_column_no_pk (
       C_CUSTKEY BIGINT,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'column'
   );

   行指向ストレージ (プライマリキーあり)

   CREATE TABLE test_table_row (
       C_CUSTKEY BIGINT PRIMARY KEY,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'row'
   );

   行指向ストレージ (プライマリキーなし)

   CREATE TABLE test_table_row_no_pk (
       C_CUSTKEY BIGINT,
       C_NAME TEXT,
       C_ADDRESS TEXT,
       C_NATIONKEY INT,
       C_PHONE TEXT,
       C_ACCTBAL DECIMAL(15, 2),
       C_MKTSEGMENT TEXT,
       C_COMMENT TEXT
   )
   WITH (
       orientation = 'row'
   );

テスト設定

このトピックでは、さまざまなモードでのインポートパフォーマンスをテストします。

1. EMR Spark クラスターのマスターノードにログオンします。詳細については、「ECS インスタンスへの接続」をご参照ください。ダウンロードした Spark-Connector パッケージをアップロードし、次のコマンドを実行して Spark SQL 対話型シェルに入ります。

   説明

   この例では、spark.executor.instances パラメーターによって同時実行数が 40 に設定されています。読み取りの並列処理をさらに細かく制御するには、spark.sql.files.maxPartitionBytes パラメーターで HDFS ファイルのパーティションサイズを調整します。

   # Spark SQL 対話型シェルに入ります。
   spark-sql --jars <path>/hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
   --conf spark.executor.instances=40 \
   --conf spark.executor.cores=1 \
   --conf spark.executor.memory=4g \
   --conf spark.sql.files.maxPartitionBytes=644245094

   コマンドの path は、hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar ファイルを含むディレクトリです。

2. Spark SQL 対話型シェルで、次の SQL 文を実行して一時ビューを作成し、データを書き込みます。

   このテストでは、さまざまなモードでの書き込みパフォーマンスを評価する際のパラメーター調整を容易にするために、一時ビューを使用します。

   説明

   実際には、カタログを使用して Hologres テーブルを直接ロードすることもできます。

   -- CSV 形式の一時ビューを作成します。
   CREATE TEMPORARY VIEW csvtable (
     c_custkey BIGINT,
     c_name STRING,
     c_address STRING,
     c_nationkey INT,
     c_phone STRING,
     c_acctbal DECIMAL(15, 2),
     c_mktsegment STRING,
     c_comment STRING)
   USING csv OPTIONS (
     path "<spark_resources>/customer.tbl", sep "|"
   );
   CREATE TEMPORARY VIEW hologresTable (
     c_custkey BIGINT,
     c_name STRING,
     c_address STRING,
     c_nationkey INT,
     c_phone STRING,
     c_acctbal DECIMAL(15, 2),
     c_mktsegment STRING,
     c_comment STRING)
   USING hologres OPTIONS (
     jdbcurl "jdbc:postgresql://<hologres_vpc_endpoint>/<database_name>",
     username "<accesskey_id>", 
     password "<accesskey_secret>", 
     table "<table_name>",
     direct_connect "false",
     write.mode "auto",
     write.insert.thread_size "3",
     write.insert.batch_size "2048"
   );
   INSERT INTO hologresTable SELECT * FROM csvTable;

   次の表にパラメーターを示します。

   パラメーター	説明
   spark_resources	EMR Spark クラスターの作成時に設定する[クラスターのルートストレージパス]です。 このパスは、EMR on ECS コンソールのクラスターの Basic Information ページの [Cluster Information] セクションにあります。
   hologres_vpc_endpoint	Hologres インスタンスの VPC ドメイン名 です。 Hologres コンソールにログオンし、ターゲットインスタンスの ID をクリックして、インスタンス詳細ページのネットワーク情報セクションで指定された VPC のエンドポイントを見つけます。たとえば、中国 (杭州) リージョンの VPC エンドポイントの形式は <Instance ID>-cn-hangzhou-vpc-st.hologres.aliyuncs.com:80 です。
   database_name	Hologres インスタンス内のデータベースの名前です。
   accesskey_id	指定された Hologres データベースへの書き込み権限を持つ AccessKey ID です。
   accesskey_secret	指定された Hologres データベースへの書き込み権限を持つ AccessKey Secret です。
   table_name	宛先の Hologres テーブルの名前です。
   write.mode	書き込みモードです。有効な値： auto ：デフォルト値。コネクターが最適なモードを自動的に選択します。 insert ：INSERT INTO VALUES 文を使用してデータを書き込みます。 stream ：FIXED COPY を使用してストリーミング書き込みを実行します。 bulk_load ：COPY を使用して、プライマリキーのないテーブルに一括インポートを実行します。 bulk_load_on_conflict ：COPY を使用して、プライマリキーのあるテーブルに一括インポートを実行します。
   write.insert.thread_size	書き込み同時実行数です。このパラメーターは insert モードにのみ適用されます。
   write.insert.batch_size	書き込みバッチサイズです。このパラメーターは insert モードにのみ適用されます。
   write.on_conflict_action	INSERT_OR_REPLACE (デフォルト)：プライマリキーの競合が発生した場合、データは更新されます。 INSERT_OR_IGNORE ：プライマリキーの競合が発生した場合、データは無視されます。

   パラメーターの詳細については、「パラメーター」をご参照ください。

テストシナリオ

テスト結果

テスト結果には次のフィールドが含まれます。

spark-sql (default)> insert into hologresTable select * from csvTable;
25/04/11 15:05:51 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:05:52 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:05:52 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:02 WARN [main] PGStream: create socket to
25/04/11 15:06:02 WARN [main] JDBCUtil$: could not connect directly
25/04/11 15:06:03 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:04 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:04 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:05 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:05 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:06 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:06 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:07 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:07 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:09 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:09 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:10 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:10 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:11 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:11 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:12 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:12 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
25/04/11 15:06:13 WARN [embedded-hologres-connector-spark-SparkSQL::xxx -worker]
Time taken: 228.901 seconds

SELECT
    COUNT(*), AVG(duration) AS avg_duration_ms
FROM
    hologres.hg_query_log
WHERE
    query_start >= '<start_time>' AND query_start <= '<end_time>'
    AND query LIKE '%<hologres_table_name>%' AND command_tag = 'COPY';

次の表は、詳細なテスト結果を示しています。