E-MapReduce:Lire et écrire dans StarRocks

Écrivez des données dans StarRocks à l'aide d'EMR Serverless Spark.

1. Créez une session SQL. Pour plus d'informations, consultez Gérer les sessions SQL.

   Lors de la création d'une session, sélectionnez la version de moteur correspondant à la version du connecteur Spark, choisissez la connexion réseau créée à l'étape précédente dans Network Connection, et ajoutez les paramètres suivants dans Spark Configuration pour charger le connecteur Spark.

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

   Remplacez oss://<bucketname>/path/connector.jar par le chemin OSS du connecteur Spark que vous avez importé à l'étape 1. Par exemple : oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar.

2. Sur la page Development, créez une tâche SparkSQL. Ensuite, sélectionnez la session SQL que vous venez de créer dans le coin supérieur droit.

   Pour plus d'informations, consultez Développer une job SparkSQL.

3. Copiez le code suivant dans le nouvel onglet SparkSQL, remplacez les valeurs des espaces réservés selon vos besoins, puis cliquez sur 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);

   Description des paramètres :

   • <fe_host> : le point de terminaison interne ou public du nœud FE de votre instance EMR Serverless StarRocks. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

     • Si vous utilisez un point de terminaison interne, assurez-vous que les deux services se trouvent dans le même VPC.

     • Si vous utilisez un point de terminaison public, vérifiez que les règles du groupe de sécurité autorisent le trafic sur les ports requis. Pour plus d'informations, consultez Paramètres d'accès réseau et de sécurité.

   • <fe_http_port> : le port HTTP du nœud FE de l'instance EMR Serverless StarRocks. La valeur par défaut est 8030. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

   • <fe_query_port> : le port de requête du nœud FE de l'instance EMR Serverless StarRocks. La valeur par défaut est 9030. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

   • <user> : le nom d'utilisateur de l'instance Serverless StarRocks. Par défaut, un utilisateur administrateur disposant des privilèges d'administration est fourni. Vous pouvez également ajouter de nouveaux utilisateurs pour vous connecter via la page User Management. Pour savoir comment ajouter des utilisateurs, consultez Gérer les utilisateurs et l'autorisation des données.

   • <password> : le mot de passe correspondant à l'utilisateur <user>.

Interrogez les données écrites à l'aide d'EMR Serverless Spark.

Dans cet exemple, nous créons une vue temporaire nommée test_view dans la tâche Spark SQL et l'utilisons pour interroger les données de score_board. Copiez le code suivant dans un nouvel onglet Spark SQL, sélectionnez-le, puis cliquez sur 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;

Résultat

La requête renvoie deux enregistrements : (1, "starrocks", 100) et (2, "spark", 100).

Écrivez des données dans StarRocks à l'aide d'EMR Serverless Spark.

1. Créez une session Notebook. Pour plus d'informations, consultez Gérer les sessions Notebook.

   Lors de la création d'une session, sélectionnez la version de moteur correspondant à la version du connecteur Spark, choisissez la connexion réseau créée à l'étape précédente dans Network Connection, et ajoutez les paramètres suivants dans Spark Configuration pour charger le connecteur Spark.

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

   Remplacez oss://<bucketname>/path/connector.jar par le chemin OSS du connecteur Spark que vous avez importé à l'étape 1. Par exemple : oss://emr-oss/spark/starrocks-spark-connector-3.5_2.12-1.1.2.jar.

2. Sur la page Development, créez une tâche Interactive Development > Notebook, puis sélectionnez la session Notebook créée dans le coin supérieur droit.

   Pour plus d'informations, consultez Gérer les sessions Notebook.

3. Copiez le code suivant dans une cellule Python du Notebook et cliquez sur Run.

   # Replace the placeholders with your EMR Serverless StarRocks configurations.
   fe_host = "<fe_host>"
   fe_http_port = "<fe_http_port>"
   fe_query_port = "<fe_query_port>"
   user = "<user>"
   password = "<password>"
   # Create a view.
   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.
   insert_data_sql = """
   INSERT INTO `score_board` VALUES (1, "starrocks", 100), (2, "spark", 100)
   """
   spark.sql(insert_data_sql)
   

   Description des paramètres :

   • <fe_host> : le point de terminaison interne ou public du nœud FE de votre instance EMR Serverless StarRocks. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

     • Si vous utilisez un point de terminaison interne, assurez-vous que les deux services se trouvent dans le même VPC.

     • Si vous utilisez un point de terminaison public, vérifiez que les règles du groupe de sécurité autorisent le trafic sur les ports requis. Pour plus d'informations, consultez Paramètres d'accès réseau et de sécurité.

   • <fe_http_port> : le port HTTP du nœud FE de l'instance EMR Serverless StarRocks. La valeur par défaut est 8030. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

   • <fe_query_port> : le port de requête du nœud FE de l'instance EMR Serverless StarRocks. La valeur par défaut est 9030. Vous trouverez cette information sur la page Instance Details, dans la section FE Details.

   • <user> : le nom d'utilisateur de l'instance Serverless StarRocks. Par défaut, un utilisateur administrateur disposant des privilèges d'administration est fourni. Vous pouvez également ajouter de nouveaux utilisateurs pour vous connecter via la page User Management. Pour savoir comment ajouter des utilisateurs, consultez Gérer les utilisateurs et l'autorisation des données.

   • <password> : le mot de passe correspondant à l'utilisateur <user>.

Interrogez les données écrites à l'aide d'EMR Serverless Spark.

Dans une nouvelle cellule Python, créez une vue temporaire nommée test_view pour interroger la table score_board. Copiez le code suivant dans la cellule et cliquez sur l'icône d'exécution () pour l'exécuter.

# Create a 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 the data.
query_sql="SELECT * FROM test_view"
result_df = spark.sql(query_sql)
result_df.show()

Résultat

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