E-MapReduce:StarRocks への読み取りと書き込み

EMR Serverless Spark を使用して StarRocks にデータを書き込みます。

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

   セッションを作成する際、Spark コネクタのバージョンに対応するエンジンバージョンを選択し、Network Connection で前のステップで作成したネットワーク接続を選択し、Spark Configuration に以下のパラメーターを追加して、Spark コネクタをロードします。

   spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar

   oss://<bucketname>/path/connector.jar を、ステップ1でアップロードした Spark コネクタの OSS パスに置き換えます。例：oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar

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

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

3. 新しい SparkSQL タブに次のコードをコピーし、必要に応じてプレースホルダーの値を置き換えてから、Run をクリックします。

   CREATE TEMPORARY VIEW score_board
   USING starrocks
   OPTIONS
   (
     "starrocks.table.identifier" = "testdb.score_board",
     "starrocks.fe.http.url" = "<fe_host>:<fe_http_port>",
     "starrocks.fe.jdbc.url" = "jdbc:mysql://<fe_host>:<fe_query_port>",
     "starrocks.user" = "<user>",
     "starrocks.password" = "<password>"
   );
   INSERT INTO `score_board` VALUES (1, "starrocks", 100), (2, "spark", 100);

   パラメーターの説明：

   • <fe_host>：EMR Serverless StarRocks インスタンスの FE ノードの内部またはパブリックエンドポイント。これは、Instance Details ページの FE 詳細 セクションで確認できます。

     • 内部エンドポイントを使用する場合、両方のサービスが同じ VPC 内にあることを確認してください。

     • パブリックエンドポイントを使用する場合、セキュリティグループのルールが必要なポートでのトラフィックを許可していることを確認してください。詳細については、「ネットワークアクセスとセキュリティ設定」をご参照ください。

   • <fe_http_port>：EMR Serverless StarRocks インスタンスの FE ノードの HTTP ポート。デフォルトは 8030 です。これは、Instance Details ページの FE 詳細 セクションで確認できます。

   • <fe_query_port>：EMR Serverless StarRocks インスタンスの FE ノードのクエリポート。デフォルトは 9030 です。これは、Instance Details ページの FE 詳細 セクションで確認できます。

   • <user>：Serverless StarRocks インスタンスのユーザー名。デフォルトでは、管理者権限を持つ admin ユーザーが提供されます。User Management ページを使用して新しいユーザーを追加して接続することもできます。ユーザーの追加方法の詳細については、「ユーザーとデータ権限の管理」をご参照ください。

   • <password>：<user> に対応するパスワード。

EMR Serverless Spark を使用して書き込まれたデータをクエリします。

この例では、Spark SQL タスクで test_view という名前の一時ビューを作成して score_board からデータをクエリします。新しい Spark SQL タブに次のコードをコピーし、コードを選択してから Run the selected をクリックします。

CREATE TEMPORARY VIEW test_view
USING starrocks
OPTIONS
(
   "starrocks.table.identifier" = "testdb.score_board",
   "starrocks.fe.http.url" = "<fe_host>:<fe_http_port>",
   "starrocks.fe.jdbc.url" = "jdbc:mysql://<fe_host>:<fe_query_port>",
   "starrocks.user" = "<user>",
   "starrocks.password" = "<password>"
);
SELECT * FROM test_view;

出力

クエリは 2 つのレコード (1, "starrocks", 100) と (2, "spark", 100) を返します。

EMR Serverless Spark を使用して StarRocks にデータを書き込みます。

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

   セッションを作成する際、Spark コネクタのバージョンに対応するエンジンバージョンを選択し、Network Connection で前のステップで作成したネットワーク接続を選択し、Spark Configuration に以下のパラメーターを追加して、Spark コネクタをロードします。

   spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar

   oss://<bucketname>/path/connector.jar を、ステップ1でアップロードした Spark コネクタの OSS パスに置き換えます。例：oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar

2. Development ページで、[Interactive Development] > [Notebook] タスクを作成し、右上隅で作成した Notebook セッションを選択します。

   詳細については、「Notebook セッションの管理」をご参照ください。

3. Notebook の Python セルに次のコードをコピーし、Run をクリックします。

   # プレースホルダーを EMR Serverless StarRocks の設定に置き換えます。
   fe_host = "<fe_host>"
   fe_http_port = "<fe_http_port>"
   fe_query_port = "<fe_query_port>"
   user = "<user>"
   password = "<password>"
   # ビューを作成します。
   create_table_sql = f"""
   CREATE TEMPORARY VIEW score_board
   USING starrocks
   OPTIONS (
     "starrocks.table.identifier" = "testdb.score_board",
     "starrocks.fe.http.url" = "{fe_host}:{fe_http_port}",
     "starrocks.fe.jdbc.url" = "jdbc:mysql://{fe_host}:{fe_query_port}",
     "starrocks.user" = "{user}",
     "starrocks.password" = "{password}"
   )
   """
   spark.sql(create_table_sql)
   # データを挿入します。
   insert_data_sql = """
   INSERT INTO `score_board` VALUES (1, "starrocks", 100), (2, "spark", 100)
   """
   spark.sql(insert_data_sql)
   

   パラメーターの説明：

   • <fe_host>：EMR Serverless StarRocks インスタンスの FE ノードの内部またはパブリックエンドポイント。これは、Instance Details ページの FE 詳細 セクションで確認できます。

     • 内部エンドポイントを使用する場合、両方のサービスが同じ VPC 内にあることを確認してください。

     • パブリックエンドポイントを使用する場合、セキュリティグループのルールが必要なポートでのトラフィックを許可していることを確認してください。詳細については、「ネットワークアクセスとセキュリティ設定」をご参照ください。

   • <fe_http_port>：EMR Serverless StarRocks インスタンスの FE ノードの HTTP ポート。デフォルトは 8030 です。これは、Instance Details ページの FE 詳細 セクションで確認できます。

   • <fe_query_port>：EMR Serverless StarRocks インスタンスの FE ノードのクエリポート。デフォルトは 9030 です。これは、Instance Details ページの FE 詳細 セクションで確認できます。

   • <user>：Serverless StarRocks インスタンスのユーザー名。デフォルトでは、管理者権限を持つ admin ユーザーが提供されます。User Management ページを使用して新しいユーザーを追加して接続することもできます。ユーザーの追加方法の詳細については、「ユーザーとデータ権限の管理」をご参照ください。

   • <password>：<user> に対応するパスワード。

EMR Serverless Spark を使用して書き込まれたデータをクエリします。

新しい Python セルで、test_view という名前の一時ビューを作成して score_board テーブルをクエリします。次のコードをセルにコピーし、実行アイコン () をクリックして実行します。

# ビューを作成します。
create_view_sql=f"""
CREATE TEMPORARY VIEW test_view
USING starrocks
OPTIONS (
  "starrocks.table.identifier" = "testdb.score_board",
  "starrocks.fe.http.url" = "{fe_host}:{fe_http_port}",
  "starrocks.fe.jdbc.url" = "jdbc:mysql://{fe_host}:{fe_query_port}",
  "starrocks.user" = "{user}",
  "starrocks.password" = "{password}"
)
"""
spark.sql(create_view_sql)
# データをクエリします。
query_sql="SELECT * FROM test_view"
result_df = spark.sql(query_sql)
result_df.show()

出力

+---+---------+-----+
| id|     name|score|
+---+---------+-----+
|  2|    spark|  100|
|  1|starrocks|  100|
+---+---------+-----+