开源大数据平台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|
+---+---------+-----+