:Spark を使用してデータをインポートする

前提条件

• Hologres インスタンスのバージョンが V0.9 以降であること。 Hologres インスタンスのバージョンは、Hologres コンソールのインスタンス詳細ページで確認できます。 Hologres インスタンスのバージョンが V0.9 より前の場合は、Hologres コンソールで Hologres インスタンスを手動でアップグレードするか、テクニカル サポートを受けるために Hologres の DingTalk グループに参加してください。 Hologres コンソールで Hologres インスタンスを手動でアップグレードする方法の詳細については、インスタンスのアップグレード をご参照ください。テクニカル サポートを受ける方法の詳細については、Hologres のオンライン サポートを受ける をご参照ください。

• Hologres でサポートされているバージョンの Spark 環境がインストールされていること。 これにより、Spark 環境で spark-shell コマンドを実行できます。

接続の使用

Hologres の Spark コネクタを使用してデータの読み取りまたは書き込みを行う場合、Java Database Connectivity（JDBC）接続が使用されます。 使用される接続数は、以下の項目の影響を受けます。

• 並列 Spark タスクの数: ジョブの実行中に Spark UI で数を確認できます。

• データ書き込みモード: Spark コネクタを使用して固定コピー モードでデータを書き込む場合、各並列タスクは 1 つの JDBC 接続を使用します。 INSERT ステートメントを使用してデータを書き込む場合、各並列タスクで使用される JDBC 接続の数は write_thread_size の値と同じです。 データ読み取りの各並列タスクは 1 つの JDBC 接続を使用します。

• その他: ジョブの開始時に、スキーマ情報を取得する必要がある場合があります。 この場合、JDBC 接続が短時間使用される可能性があります。

ジョブで使用される JDBC 接続の総数は、次の式を使用して計算できます。

• 固定コピー モード: Parallelism × 1 + 1

• INSERT ステートメント: Parallelism × write_thread_size + 1

(推奨) Spark コネクタを使用して Hologres にデータを書き込む

Hologres の組み込み Spark コネクタを使用して Hologres にデータを書き込むことをお勧めします。 Spark コネクタは、Holo Client と一緒に使用されます。 他の書き込み方法と比較して、Spark コネクタはより優れた書き込みパフォーマンスを提供します。 Spark コネクタを使用してデータを書き込むには、次の手順を実行します。 サンプル コードの詳細については、このトピックの Spark コネクタを使用して Hologres にデータを書き込む例 セクションをご参照ください。

Spark を使用して特定のタイプのソースからデータを読み取り、Hologres に書き込む

1. 特定のタイプのソースからデータを読み取ります。

   Spark を使用して、さまざまなタイプのソースからデータを読み取ることができます。 次の例は、Hologres または別のタイプのソースからデータを読み取る方法を示しています。

   • Hologres からデータを読み取る

     Hologres は PostgreSQL と互換性があります。 PostgreSQL JDBC ドライバに基づいて Spark を使用して Hologres データを読み取ることができます。 次のサンプル コードは参照用です。

     説明

     Hologres データを読み取る前に、公式 Web サイトで PostgreSQL JDBC JAR パッケージをダウンロードしてください。 この例では、postgresql-42.2.18 が使用されています。 次に、spark-shell で ./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar コマンドを実行して、PostgreSQL JDBC JAR パッケージを読み込みます。 Hologres の Spark コネクタの JAR パッケージと一緒に PostgreSQL JDBC JAR パッケージを読み込むこともできます。

     // 例えば tb008 などのテーブルから読み取る
     val readDf = spark.read
       .format("jdbc") // PostgreSQL JDBC ドライバに基づいて Hologres データを読み取ります。
       .option("driver","org.postgresql.Driver")
       .option("url", "jdbc:postgresql://Ip:Por/test_database")
       .option("dbtable", "tb008")
       .option("user", "your_username")
       .option("password", "your_password")
       .load()

     Spark connector V1.3.2 以降では、Hologres からデータを読み取り、並列データ読み取りのパフォーマンスを最適化できます。 PostgreSQL JDBC ドライバと比較して、Spark コネクタではデータ読み取りの並列処理を設定でき、Hologres でのデータシャーディングをサポートして並列データ読み取りを実装できます。 これにより、パフォーマンスが大幅に向上します。 サンプルコード:

     val spark = SparkSession
     .builder
     .appName("ReadFromHologres")
     .master("local[*]")
     .getOrCreate()
     
     spark.sparkContext.setLogLevel("WARN")
     
     import spark.implicits._
     
     val schema = StructType(Array(
       StructField("id", LongType),
       StructField("counts", IntegerType),
       StructField("name", StringType, false),
       StructField("price", DecimalType(38, 12)),
       StructField("out_of_stock", BooleanType)
     ))
     
     val readDf = spark.read
     .format("hologres")
     .schema(schema) // オプション。 スキーマを指定しない場合、デフォルトでは Hologres テーブルのすべてのフィールドが読み取られます。
     .option("username", "your_username")
     .option("password", "your_password")
     .option("jdbcurl", "jdbc:postgresql://hologres_endpoint/test_db")
     .option("table", "tb008")
     .option("scan_parallelism", "10") // Hologres からデータを読み取るためのデフォルトの並列処理。 最大値は、Hologres テーブルのシャード数です。
     .load()

   • Parquet ファイルなどの別のタイプのソースからデータを読み取る

     Spark を使用して、Parquet ファイルや Hive テーブルなど、他のタイプのソースからデータを読み取ることができます。 次の例は、サンプル コードを提供します。

     import org.apache.spark.{SparkConf, SparkContext}
     import org.apache.spark.sql.hive.HiveContext
     
     val sparkConf = new SparkConf()
     val sc = new SparkContext(sparkConf)
     val hiveContext = new HiveContext(sc)
     
     // 例えば phone などのテーブルから読み取る
     val readDf = hiveContext.sql("select * from hive_database.phone")

2. データを Hologres に書き込みます。

   import com.alibaba.hologres.spark2.sink.SourceProvider
   
   -- hologres テーブルに書き込む
   df.write
     .format("hologres")
     .option(SourceProvider.USERNAME, "your_username")
     .option(SourceProvider.PASSWORD, "your_password")
     .option(SourceProvider.ENDPOINT, "Ip:Port")
     .option(SourceProvider.DATABASE, "test_database")
     .option(SourceProvider.TABLE, table)
     .option(SourceProvider.WRITE_BATCH_SIZE, 512) -- バッチで許可されるリクエストの最大数。
     .option(SourceProvider.INPUT_DATA_SCHEMA_DDL, df.schema.toDDL) -- このパラメータは Spark 3.X の場合にのみ必要です。
     .mode(SaveMode.Append) // このパラメータは Spark 3.X の場合にのみ必要です。
     .save()

Spark を使用してリアルタイムで Hologres にデータを書き込む

1. 次のステートメントを実行して、Hologres にデータを書き込むテーブルを作成します。

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. オンプレミス マシンからデータを読み取ります。 単語頻度の統計を収集し、統計をリアルタイムで Hologres に書き込みます。 次の例は、サンプル コードを提供します。

   • コード

      val spark = SparkSession
           .builder
           .appName("StreamToHologres")
           .master("local[*]")
           .getOrCreate()
     
         spark.sparkContext.setLogLevel("WARN")
         import spark.implicits._
     
         val lines = spark.readStream
           .format("socket")
           .option("host", "localhost")
           .option("port", 9999)
           .load()
     
         -- 行を単語に分割する
         val words = lines.as[String].flatMap(_.split(" "))
     
         -- 実行中の単語数を生成する
         val wordCounts = words.groupBy("value").count()
     
         wordCounts.writeStream
             .outputMode(OutputMode.Complete())
             .format("hologres")
             .option(SourceProvider.USERNAME, "your_username")
             .option(SourceProvider.PASSWORD, "your_password")
             .option(SourceProvider.JDBCURL, "jdbc:postgresql://Ip:Port/dbname")
             .option(SourceProvider.TABLE, "test_table_stream")
             .option("batchsize", 1)
             .option("isolationLevel", "NONE")
             .option("checkpointLocation", checkpointLocation)
             .start()
             .awaitTermination()

   • パラメータ

     パラメータ	デフォルト値	必須	説明
     username	デフォルト値なし	はい	Alibaba Cloud アカウントの AccessKey ID。セキュリティ管理 ページで AccessKey ID を取得できます。 環境変数を設定し、環境変数から AccessKey ID と AccessKey シークレットを取得することをお勧めします。 これにより、リークのリスクを軽減できます。
     password	デフォルト値なし	はい	Alibaba Cloud アカウントの AccessKey シークレット。セキュリティ管理 環境変数を設定し、環境変数から AccessKey ID と AccessKey シークレットを取得することをお勧めします。 これにより、リークのリスクを軽減できます。
     table	デフォルト値なし	はい	データを書き込む Hologres テーブルの名前。
     endpoint	デフォルト値なし	このパラメータまたは JDBCURL パラメータを設定します。	Hologres インスタンスのエンドポイント。 Hologres コンソール のインスタンス詳細ページの [ネットワーク情報] セクションで、Hologres インスタンスのエンドポイントを取得できます。
     database	デフォルト値なし	このパラメータまたは JDBCURL パラメータを設定します。	宛先テーブルが存在する Hologres データベースの名前。
     jdbcurl	デフォルト値なし	このパラメータのみ、または ENDPOINT パラメータと DATABASE パラメータを設定します。	Hologres インスタンスの JDBC URL。
     copy_write_mode	true	いいえ	固定コピー モードでデータを書き込むかどうかを指定します。 固定コピーは、Hologres V1.3 でサポートされている新機能です。 固定コピー モードでは、データはバッチではなくストリーミング モードで書き込まれます。 したがって、固定コピー モードでのデータ書き込みは、INSERT ステートメントを使用したデータ書き込みよりもスループットが高く、データ レイテンシが低く、クライアント メモリ リソースの消費が少なくなります。 説明 固定コピー モードを使用するには、コネクタのバージョンが V1.3.0 以降であり、Hologres エンジンのバージョンが V1.3.34 以降である必要があります。
     copy_write_format	false	いいえ	不正なデータを確認するかどうかを指定します。 このパラメータは、COPY_WRITE_MODE が true に設定されている場合にのみ有効になります。 このパラメータを true に設定し、不正なデータが生成された場合、システムは書き込みに失敗した行を特定できます。 説明 不正なデータの確認は、書き込みパフォーマンスに悪影響を及ぼします。 トラブルシューティング プロセスでのみ、このパラメータを true に設定することをお勧めします。
     bulk_load	true	いいえ	バッチ コピー モードで Hologres にデータを書き込むかどうかを指定します。 固定コピー モードでは、データはストリーミング モードで書き込まれます。 説明 Hologres V2.1 では、プライマリ キーのないテーブルへのデータ書き込みのパフォーマンスが最適化されています。 Hologres V2.1 では、バッチ モードでプライマリ キーのないテーブルにデータを書き込む操作は、テーブル ロックを取得するのではなく、行ロックを取得します。 これにより、バッチ モードでのデータ書き込みを、固定プランを使用して実行される操作と並行して実行できます。 これにより、データ処理の効率と並列処理が向上します。 コネクタのバージョンが V1.4.0 以降の場合は、Hologres エンジンのバージョンが V2.1.0 以降である必要があります。
     max_cell_buffer_size	20971520 (20 MB)	いいえ	COPY_WRITE_MODE が true に設定されている場合のフィールドの最大長。
     copy_write_dirty_data_check	false	いいえ	不正なデータを確認するかどうかを指定します。 このパラメータを true に設定し、不正なデータが生成された場合、システムは書き込みに失敗した行を特定できます。 不正なデータの確認は、書き込みパフォーマンスに悪影響を及ぼします。 トラブルシューティング プロセスでのみ、このパラメータを true に設定することをお勧めします。 説明 このパラメータは、COPY_WRITE_MODE が true に設定されている場合にのみ有効になります。
     copy_write_direct_connect	直接接続が許可されているシナリオでは true	いいえ	このパラメータは、COPY_WRITE_MODE が true に設定されている場合にのみ有効になります。 コピー モードで書き込むことができるデータ量は、VPC エンドポイントのスループットに基づいて決定されます。 システムは、コピー モードでデータが書き込まれるときに、環境が Hologres FE ノードに直接接続できるかどうかを確認します。 環境が Hologres FE ノードに直接接続できる場合は、デフォルトで直接接続が使用されます。 このパラメータを false に設定すると、直接接続は使用されません。
     input_data_schema_ddl	デフォルト値なし	このパラメータは Spark 3.X で必要です。 <your_DataFrame>.schema.toDDL 形式でパラメータ値を設定します。	Spark の DataFrame の DDL ステートメント。
     write_mode	INSERT_OR_REPLACE	いいえ	プライマリ キーの競合を処理するために使用されるポリシー。 宛先テーブルにプライマリ キーがある場合は、このパラメータが必要です。 有効な値: INSERT_OR_IGNORE: プライマリ キーの競合が発生した場合、書き込もうとしているデータを破棄します。 INSERT_OR_UPDATE: プライマリ キーの競合が発生した場合、宛先テーブルの関連列を更新します。 INSERT_OR_REPLACE: プライマリ キーの競合が発生した場合、宛先テーブルのすべての列を更新します。
     write_batch_size	512	いいえ	データを書き込むスレッドのバッチで許可されるリクエストの最大数。 write_mode パラメータに基づいて競合が処理された後、PUT リクエストの総数が write_batch_size パラメータで指定された上限に達すると、データはバッチで送信されます。
     write_batch_byte_size	2 MB	いいえ	データを書き込むスレッドのバッチで許可されるバイトの最大数。 WRITE_MODE パラメータに基づいて競合が処理された後、PUT リクエストのバイトの総数が WRITE_BATCH_BYTE_SIZE パラメータで指定された上限に達すると、データはバッチで送信されます。
     write_max_interval_ms	10000 ms	いいえ	データがバッチで送信される間隔。
     write_fail_strategy	TYR_ONE_BY_ONE	いいえ	送信エラーを処理するために使用されるポリシー。 バッチを送信できない場合、Holo Client はバッチ内のデータエントリを指定された順序で 1 つずつ送信します。 データエントリを送信できない場合、Holo Client はデータエントリに関する情報を含むエラーメッセージを返します。
     write_thread_size	1	いいえ	データの書き込みに使用される並列スレッドの数。 各スレッドは 1 つの接続を占有します。 Spark ジョブによって占有される接続の総数は、Spark の並列処理によって異なります。 接続の総数は、次の式を使用して計算できます: 接続の総数 = spark.default.parallelism の値 × WRITE_THREAD_SIZE の値。
     dynamic_partition	false	いいえ	パーティションのない親テーブルにデータが書き込まれた場合に、パーティションを自動的に作成するかどうかを指定します。 有効な値: true および false。 true: パーティションのない親テーブルにデータが書き込まれた場合、パーティションが自動的に作成されます。
     retry_count	3	いいえ	接続エラーが発生した場合に、データの書き込みとクエリに許可される再試行の最大回数。
     retry_sleep_init_ms	1000 ms	いいえ	リクエストの再試行に費やされる時間は、次の式を使用して計算されます: retry_sleep_init_ms + retry_count × retry_sleep_step_ms。
     retry_sleep_step_ms	10*1000 ms	いいえ	リクエストの再試行に費やされる時間は、次の式を使用して計算されます: retry_sleep_init_ms + retry_count × retry_sleep_step_ms。
     connection_max_idle_ms	60000 ms	いいえ	データの読み取りと書き込みに使用される接続のアイドル タイムアウト期間。 接続が指定されたアイドル タイムアウト期間よりも長くアイドル状態のままである場合、Holo Client は接続を自動的に解放します。
     fixed_connection_mode	false	いいえ	固定接続を使用するかどうかを指定します。 固定コピー以外のモード（INSERT モードなど）では、データの書き込みとポイント クエリは接続を占有しません。 説明 固定接続機能はベータ版であり、コネクタのバージョンが V1.2.0 以降、Hologres エンジンのバージョンが V1.3.0 以降の場合にのみ使用できます。
     scan_batch_size	256	いいえ	Hologres からデータを読み取るときに毎回スキャンされる行数。
     scan_timeout_seconds	60	いいえ	Hologres からデータを読み取るときのスキャン操作のタイムアウト期間。 単位: 秒。
     scan_parallelism	10	いいえ	Hologres からデータを読み取るために使用されるシャードの数。 最大値は、Hologres テーブルのシャード数です。 ジョブの実行時、シャードはデータ読み取りのために Spark タスクに割り当てられます。

データ型マッピング

次の表は、Spark と Hologres 間のデータ型マッピングを示しています。