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;

出力

クエリにより、(1, "starrocks", 100) と (2, "spark", 100) の 2 つのレコードが返されます。

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

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

   セッションを作成する際、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] ジョブを作成し、右上で作成したノートブックセッションを選択します。

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

3. 次のコードをノートブックの 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|
+---+---------+-----+