E-MapReduce:Hologres のデータの読み取りと書き込み

例1：SQL セッション

SQL セッションを使用して Hologres のデータの読み取りと書き込みを行います。

1. SQL セッションを作成します。詳細については、「SQLセッションの管理」をご参照ください。

   セッションを作成するときは、前の手順で作成したネットワーク接続をネットワーク接続リストから選択します。 Spark Configuration セクションで、hologres-connector-spark をロードするための次のパラメーターを追加します。

   # hologres-connector JAR ファイルを追加します (esr-4.8.0 より前のバージョンでのみ必要です)。
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar
   
   # Hologres カタログを設定します。
   spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog
   spark.sql.catalog.hologres_external_test_db.username ***
   spark.sql.catalog.hologres_external_test_db.password ***
   spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb

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

   パラメーター	例	説明
   spark.emr.serverless.user.defined.jars	oss://<bucket>/hologres-connector-spark-3.x-<version>.jar	ユーザー定義 JAR ファイルへのパス。
   spark.sql.catalog.hologres_external_test_db	com.alibaba.hologres.spark3.HoloTableCatalog	Hologres データソースを Spark 3.x の外部カタログとして設定します。これは固定値です。
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	お使いの Alibaba Cloud アカウントの AccessKey ID。機密情報の処理にはシークレット管理を使用することを推奨します。詳細については、「シークレット管理を使用した機密情報の管理」をご参照ください。
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	お使いの Alibaba Cloud アcount の AccessKey Secret。機密情報の処理にはシークレット管理を使用することを推奨します。詳細については、「シークレット管理を使用した機密情報の管理」をご参照ください。
   spark.sql.catalog.hologres_external_test_db.jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	Hologres インスタンスの JDBC 接続 URL。

   パラメーター名の中の hologres_external_test_db の部分はカスタマイズできます。

2. [Data Development] ページで [SparkSQL] ジョブを作成し、右上隅から作成した SQL セッションを選択します。

   詳細については、「SparkSQL開発」をご参照ください。

3. 次のコードを新しい SparkSQL タブにコピーし、Run をクリックします。

   -- testdb データベースに切り替えます。
   USE hologres_external_test_db;
   -- データを書き込みます。
   INSERT INTO `public`.test VALUES ('1004','tom');
   -- データをクエリします。
   SELECT * FROM `public`.test;

   

例2：ストリーミングジョブ

この PySpark の例では、Kafka からデータを読み取り、ストリーミングジョブとして Hologres に書き込みます。

1. このコード例では、Kafka の情報と Hologres のテーブルを実際の値に置き換えます。

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import col
   
   # Kafka 情報を設定します。
   servers = "alikafka-serverless-cn-xxxxx-vpc.alikafka.aliyuncs.com:9092"  # ご使用の Kafka ブートストラップサーバーに置き換えます。
   topic = "topic-name"  # ご使用の Kafka トピックに置き換えます。
   
   # SparkSession を作成します。
   spark = SparkSession.builder \
       .appName("test read kafka") \
       .getOrCreate()
   
   # Kafka ストリームを読み取ります。
   df = spark \
       .readStream \
       .format("kafka") \
       .option("kafka.bootstrap.servers", servers) \
       .option("subscribe", topic) \
       .load()
   
   # Hologres に書き込む関数を定義します (マイクロバッチごとに呼び出されます)。
   def write_to_hologres(batch_df, batch_id):
       print(f"Writing batch {batch_id} to Hologres...")
       batch_df.write \
           .format("hologres") \
           .mode("append") \
           .insertInto("hologres_external_test_db.public.test")  # ご使用の Hologres テーブルに置き換えます。
   
   # キーと値を文字列に変換してストリームを書き込みます。
   query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
       .writeStream \
       .foreachBatch(write_to_hologres) \
       .outputMode("append") \
       .trigger(processingTime='30 seconds') \
       .start()
   
   # ストリーミングクエリが終了するのを待ちます (ノートブックでは、これによりセルの実行がブロックされます)。
   query.awaitTermination()

2. ファイルをアップロードします。

   1. Artifacts ページで、Upload File をクリックします。

   2. Upload File ダイアログボックスで、アップロードエリアをクリックして前の手順の Python ファイルを選択するか、ファイルをアップロードエリアにドラッグします。

3. ストリーミングジョブを作成して実行します。

   1. Development ページで、 (作成) アイコンをクリックします。

   2. 表示されるダイアログボックスで、Name を入力し、Application (Streaming) リストから PySpark を選択して、OK をクリックします。

   3. 新しい開発タブで、次のパラメーターを設定し、その他はデフォルト設定のままにします。次に、Publish をクリックします。

      パラメーター	説明
      [Main Python Resources]	前の手順でアップロードした Python ファイルを選択します。
      Engine Version	互換性のある Spark バージョンを選択します。この例では esr-4.6.0 を使用します。
      Network Connection	ステップ2で作成したネットワーク接続を選択します。
      [Spark Configuration]	# hologres-connector JAR ファイルを追加します (esr-4.8.0 より前のバージョンでのみ必要です)。 spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar # Hologres カタログを設定します。 spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog spark.sql.catalog.hologres_external_test_db.username *** spark.sql.catalog.hologres_external_test_db.password *** spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb パラメーターの詳細な説明については、「例1：SQL セッション」をご参照ください。

   4. ジョブを公開したら、Go to O&M をクリックします。開いたページで Start をクリックします。

4. 結果を検証します。

   1. Kafka にメッセージを送信します。

      

   2. Spark SQL を使用してデータをクエリします。

例3：Notebook セッション

1. Notebook セッションを作成します。詳細については、「Notebookセッションの管理」をご参照ください。

   セッションを作成するときは、前の手順で作成したネットワーク接続をネットワーク接続リストから選択します。 Spark Configuration セクションで、hologres-connector-spark をロードするための次のパラメーターを追加します。

   # hologres-connector JAR ファイルを追加します (esr-4.8.0 より前のバージョンでのみ必要です)。
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar

2. [Data Development] ページで [Notebook] ジョブを作成し、右上隅から作成した Notebook セッションを選択します。

3. 次のコードを新しい Notebook タブにコピーし、をクリックします。

   import pandas as pd
   from pyspark.sql import SparkSession
   from pyspark.sql.types import StructType, StructField, StringType, IntegerType, LongType
   
   # 1. Pandas DataFrame を準備します。
   pdf = pd.DataFrame({
       "id": ["1006"],
       "name": ["sl"]
   })
   
   # 2. PySpark DataFrame に変換します。
   # (オプション：正しいデータ型を保証するため、スキーマを明示的に定義します)
   schema = StructType([
       StructField("id", StringType(), True),
       StructField("name", StringType(), True)
   ])
   
   df = spark.createDataFrame(pdf, schema=schema)
   
   # Hologres に書き込みます。
   df.write \
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .mode("append") \
     .save()
   
   # データを読み取ります。
   readDf = spark.read\
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .load()
   
   readDf.select("id", "name").show(10)

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

   パラメーター	例	説明
   username	LTAI******	お使いの Alibaba Cloud アカウントの AccessKey ID。機密情報の処理にはシークレット管理を使用することを推奨します。詳細については、「シークレット管理を使用した機密情報の管理」をご参照ください。
   password	mXYV******	お使いの Alibaba Cloud アカウントの AccessKey Secret。機密情報の処理にはシークレット管理を使用することを推奨します。詳細については、「シークレット管理を使用した機密情報の管理」をご参照ください。
   jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	Hologres インスタンスの JDBC 接続 URL。

4. 結果を検証します。