E-MapReduce：讀寫StarRocks

通過Serverless Spark向StarRocks寫入資料。

1. 建立SQL會話，詳情請參見管理SQL會話。

   建立會話時，選擇與Spark Connector版本對應的引擎版本，在网络连接中選擇上一步建立好的網路連接，並在Spark 配置中添加以下參數來載入Spark Connector。

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

   其中，oss://<bucketname>/path/connector.jar為您步驟一中上傳至OSS的Spark Connector的路徑。例如，oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar。

2. 在数据开发頁面，建立一個SparkSQL類型的任務，然後在右上方選擇建立好的SQL會話。

   更多操作，請參見SparkSQL開發。

3. 拷貝如下代碼到新增的SparkSQL頁簽中，並根據需要修改相應的參數資訊，然後單擊运行。

   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>：Serverless StarRocks執行個體中FE的內網或公網地址。您可以在实例详情頁面的FE 详情地區查看。

     • 如果使用內網地址，請確保在同一VPC內。

     • 如果使用公網地址，需確保安全性群組規則允許相應的連接埠通訊，詳情請參見網路訪問與安全設定。

   • <fe_http_port>：Serverless StarRocks執行個體中FE的HTTP連接埠（預設為8030）。您可以在实例详情頁面的FE 详情地區查看。

   • <fe_query_port>：Serverless StarRocks執行個體中FE的查詢連接埠（預設為9030）。您可以在实例详情頁面的FE 详情地區查看。

   • <user>：Serverless StarRocks執行個體的使用者名稱。預設提供admin使用者，具有管理員權限。您也可以通過用户管理頁面新增使用者來串連。新增使用者操作，請參見系統管理使用者及資料授權。

   • <password>：使用者 <user> 對應的密碼。

通過Serverless Spark查詢寫入的資料。

在本文樣本中，我們是在上述的SparkSQL任務中建立了一個臨時視圖 test_view，然後通過該視圖查詢 score_board 的資料。拷貝如下代碼到新增的SparkSQL頁簽中，選中代碼後單擊运行选中。

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;

返回資訊

查詢返回結果包含 id、name、score 三列，共兩行資料：id=1, name=starrocks, score=100；id=2, name=spark, score=100。

通過Serverless Spark向StarRocks寫入資料。

1. 建立Notebook會話，詳情請參見管理Notebook會話。

   建立會話時，選擇與Spark Connector版本對應的引擎版本，在网络连接中選擇上一步建立好的網路連接，並在Spark 配置中添加以下參數來載入Spark Connector。

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

   其中，oss://<bucketname>/path/connector.jar為您步驟一中上傳至OSS的Spark Connector的路徑。例如，oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar。

2. 在数据开发頁面，選擇建立一個互動式開發 > Notebook類型的任務，然後在右上方選擇建立的Notebook會話。

   更多操作，請參見管理Notebook會話。

3. 拷貝如下代碼到新增的Notebook的Python儲存格中，單擊运行。

   # 替換為您的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>：Serverless StarRocks執行個體中FE的內網或公網地址。您可以在实例详情頁面的FE 详情地區查看。

     • 如果使用內網地址，請確保在同一VPC內。

     • 如果使用公網地址，需確保安全性群組規則允許相應的連接埠通訊，詳情請參見網路訪問與安全設定。

   • <fe_http_port>：Serverless StarRocks執行個體中FE的HTTP連接埠（預設為8030）。您可以在实例详情頁面的FE 详情地區查看。

   • <fe_query_port>：Serverless StarRocks執行個體中FE的查詢連接埠（預設為9030）。您可以在实例详情頁面的FE 详情地區查看。

   • <user>：Serverless StarRocks執行個體的使用者名稱。預設提供admin使用者，具有管理員權限。您也可以通過用户管理頁面新增使用者來串連。新增使用者操作，請參見系統管理使用者及資料授權。

   • <password>：使用者 <user> 對應的密碼。

通過Serverless Spark查詢寫入的資料。

在本文樣本中，我們新增一個Python儲存格，在其中建立了一個臨時視圖 test_view，然後通過該視圖查詢 score_board 的資料。拷貝如下代碼到新增的Python儲存格中，然後單擊表徵圖。

#建立view
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|
+---+---------+-----+