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	Spark 3.x では、このパラメータを使用して Hologres のデータソースを外部カタログとして設定します。この値は固定です。
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	Alibaba Cloud アカウントの AccessKey ID です。機密情報の取り扱いには、シークレット管理の使用を推奨します。詳細については、「シークレット管理を使用した機密情報の管理」をご参照ください。
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	Alibaba Cloud アカウントの 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 をクリックします。

   -- hologres_external_test_db カタログに切り替えます。
   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 \
           .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 ページで、 (Create) アイコンをクリックします。

   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>/test_script/hologres-connector-spark-3.x-1.5.6-jar-with-dependencies.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：ノートブックセッション

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

   セッションを作成する際に、ネットワーク接続リストから前のステップで作成したネットワーク接続を選択します。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] ジョブを作成し、右上から作成したノートブックセッションを選択します。

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)

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

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

4. 結果を確認します。