Lire et écrire des données dans Doris - E-MapReduce - Centre de documentation Alibaba Cloud

EMR Serverless Spark se connecte à Doris via le connecteur Apache Doris Spark officiel. Après avoir ajouté les configurations requises, vous pouvez lire et écrire des données dans Doris depuis votre environnement Spark.

Informations générales

Apache Doris est une base de données analytique en temps réel haute performance qui prend en charge des scénarios tels que l'analyse de rapports, les requêtes ad hoc et l'accélération des requêtes fédérées sur les data lakes. Pour plus d'informations, consultez la page Présentation d'Apache Doris.

EMR Serverless Spark est une plateforme Lakehouse entièrement gérée compatible avec Spark open source. En combinant Apache Doris et EMR Serverless Spark, vous pouvez créer des flux de traitement de données de bout en bout couvrant l'ingestion, l'analyse et la diffusion des données.

Prérequis

Un espace de travail Serverless Spark a été créé. Pour plus d'informations, consultez la rubrique Créer un espace de travail.
Un cluster Doris a été créé.

Si vous créez un cluster OLAP EMR on ECS incluant le service Doris, consultez la rubrique Créer un cluster. Cette rubrique utilise comme exemple un cluster Doris créé sur EMR on ECS, appelé ci-après « cluster EMR Doris ».

Limites

La version du moteur EMR Serverless Spark doit être esr-2.6.0, esr-3.2.0, esr-4.2.0 ou une version ultérieure.

Procédure

Étape 1 : Obtenir le fichier JAR du connecteur Doris Spark et le télécharger vers OSS

Consultez la documentation officielle du connecteur Spark Doris pour connaître la matrice de compatibilité entre les versions du connecteur et les versions du moteur Spark. Assurez-vous que votre version de Spark est compatible avec la version du connecteur que vous téléchargez.

Accédez au dépôt GitHub du connecteur Doris Spark et téléchargez la version appropriée.

Le fichier Java Archive (JAR) du connecteur Doris Spark suit le format de nommage suivant : spark-doris-connector-spark-${spark_version}-${connector_version}.jar. Par exemple, si votre version de moteur est esr-3.1.0 (Spark 3.4.3, Scala 2.12), téléchargez le fichier spark-doris-connector-spark-3.4-24.0.0.jar.
Téléversez le fichier JAR du connecteur Spark téléchargé vers Alibaba Cloud OSS. Pour plus d'informations, consultez la rubrique Téléchargement simple.

Étape 2 : Créer une connexion réseau

EMR Serverless Spark nécessite une connectivité réseau avec le cluster EMR Doris. Pour plus d'informations, consultez la rubrique Connectivité réseau entre EMR Serverless Spark et d'autres VPC.

Important

Lorsque vous ajoutez des règles de groupe de sécurité, ouvrez uniquement les ports requis dans la plage Destination Port Range. Les valeurs de port valides vont de 1 à 65535. Par exemple, ouvrez le port HTTP (8031), le port RPC (9061) et le port Webserver (8041).

Étape 3 : Créer une base de données et une table dans le cluster EMR Doris

Connectez-vous au cluster via SSH. Pour plus d'informations, consultez la rubrique Se connecter à un cluster.
Exécutez la commande suivante pour vous connecter au cluster EMR Doris.
```
mysql -h127.0.0.1  -P 9031 -uroot
```

Créez une base de données et une table.

CREATE DATABASE IF NOT EXISTS testdb;
USE testdb;
CREATE TABLE test (
    id INT, 
    name STRING
) PROPERTIES("replication_num" = "1");

Insérez des données de test.

INSERT INTO test VALUES (1, 'a'), (2, 'b'), (3, 'c');

Interrogez les données.

SELECT * FROM test;

Le résultat suivant s'affiche.

MySQL [testdb]> SELECT * FROM test;
+------+------+
| id   | name |
+------+------+
|    1 | a    |
|    2 | b    |
|    3 | c    |
+------+------+
3 rows in set (0.177 sec)

Étape 4 : Lire les tables Doris à l'aide de Serverless Spark

Lire les tables Doris à l'aide d'une session SQL

Créez une session SQL. Pour plus d'informations, consultez la rubrique Gérer les sessions SQL.
Lors de la création de la session, sélectionnez la version du moteur correspondant à votre connecteur Doris Spark dans la liste déroulante Engine Version. Pour Normal Network Connection, sélectionnez la connexion créée à l'étape 2. Pour Spark Configuration, ajoutez le paramètre suivant pour charger le connecteur Doris Spark.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
Remplacez oss://<bucketname>/path/connector.jar par le chemin OSS du fichier JAR du connecteur Doris Spark que vous avez téléchargé à l'étape 1. Par exemple : oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.
Sur la page Development, créez une tâche SparkSQL, puis sélectionnez la session SQL que vous avez créée dans le coin supérieur droit.

Pour plus d'informations, consultez la rubrique Développer SparkSQL.

Copiez le code suivant dans le nouvel onglet SparkSQL, mettez à jour les paramètres selon vos besoins, puis cliquez sur Run.

CREATE TEMPORARY VIEW test
USING doris
OPTIONS(
  "table.identifier" = "testdb.test",
  "fenodes" = "<doris_address>:<http_port>",
  "user" = "<user>",
  "password" = "<password>"
);
SELECT * FROM test;

Les paramètres sont décrits ci-dessous.

Paramètre	Description	Exemple
`testdb.test`	Nom réel de la base de données et de la table dans Doris.	Si vous utilisez un autre cluster Doris, configurez les paramètres en conséquence. Si vous utilisez un cluster créé sur EMR on ECS, utilisez les valeurs suivantes : `testdb.test` : Cet exemple utilise `testdb.test`. `<doris_address>` : Dans la console EMR on ECS, accédez à la page Nodes du cluster Doris, cliquez sur l'icône à côté de emr-master et affichez l'adresse IP privée. `<http_port>` : La valeur par défaut est 8031. `<user>` : Le nom d'utilisateur par défaut est `root`. `<password>` : Le mot de passe par défaut est vide.
`<doris_address>`	Adresse IP privée du nœud exécutant le service Doris.
`<http_port>`	Numéro de port utilisé par Doris pour écouter les requêtes HTTP.
`<user>`	Nom d'utilisateur utilisé pour se connecter à Doris.
`<password>`	Mot de passe utilisé pour se connecter à Doris.

Si les données sont renvoyées avec succès, votre configuration est correcte.

Lire les tables Doris à l'aide d'une session Notebook

Créez une session Notebook. Pour plus d'informations, consultez la rubrique Gérer les sessions Notebook.
Lors de la création de la session, sélectionnez la version du moteur correspondant à votre connecteur Doris Spark dans la liste déroulante Engine Version. Sous Normal Network Connection, sélectionnez la connexion que vous avez créée à l'étape 2. Sous Spark Configuration, ajoutez le paramètre suivant pour charger le connecteur Doris Spark.
```
spark.emr.serverless.user.defined.jars  oss://<bucketname>/path/connector.jar
```
Remplacez oss://<bucketname>/path/connector.jar par le chemin OSS du fichier JAR du connecteur Doris Spark que vous avez téléchargé à l'étape 1. Par exemple : oss://emr-oss/spark/spark-doris-connector-spark-3.4-24.0.0.jar.
Sur la page Development, créez une tâche de type Interactive development > Notebook, puis sélectionnez la session Notebook que vous avez créée dans le coin supérieur droit.

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

Copiez le code suivant dans le nouvel onglet Notebook, mettez à jour les paramètres selon vos besoins, puis cliquez sur Run.

dorisSparkDF = spark.read.format("doris") \
  .option("doris.table.identifier", "testdb.test") \
  .option("doris.fenodes", "<doris_address>:<http_port>") \
  .option("user", "<user>") \
  .option("password", "<password>") \
  .load()
dorisSparkDF.show(3)

Les paramètres sont décrits ci-dessous.

Paramètre	Description	Exemple
`testdb.test`	Nom réel de la base de données et de la table dans Doris.	Si vous utilisez un autre cluster Doris, configurez les paramètres en conséquence. Si vous utilisez un cluster créé sur EMR on ECS, utilisez les valeurs suivantes : `testdb.test` : Cet exemple utilise `testdb.test`. `<doris_address>` : Dans la console EMR on ECS, accédez à la page Nodes du cluster Doris, cliquez sur l'icône à côté de emr-master et affichez l'adresse IP privée. `<http_port>` : La valeur par défaut est 8031. `<user>` : Le nom d'utilisateur par défaut est `root`. `<password>` : Le mot de passe par défaut est vide.
`<doris_address>`	Adresse IP privée du nœud exécutant le service Doris.
`<http_port>`	Numéro de port utilisé par Doris pour écouter les requêtes HTTP.
`<user>`	Nom d'utilisateur utilisé pour se connecter à Doris.
`<password>`	Mot de passe utilisé pour se connecter à Doris.

Les trois premières lignes de la table testdb.test sont renvoyées. Si les données sont renvoyées avec succès, votre configuration est correcte.

Étape 5 : Écrire dans les tables Doris à l'aide de Serverless Spark

Écrire dans les tables Doris à l'aide d'une session SQL

Copiez le code suivant dans l'onglet SparkSQL créé à l'étape précédente, mettez à jour les paramètres selon vos besoins, puis cliquez sur Run.

CREATE TEMPORARY VIEW test_write
USING doris
OPTIONS(
  "table.identifier" = "testdb.test",
  "fenodes" = "<doris_address>:<http_port>",
  "user" = "<user>",
  "password" = "<password>"
);
INSERT INTO test_write VALUES (4, 'd'), (5, 'e');
SELECT * FROM test_write;

Si les données suivantes s'affichent, l'opération d'écriture a réussi.

Le résultat attendu contient deux colonnes, id et name, avec cinq lignes. Les lignes (1, a), (2, b) et (3, c) sont les lignes d'origine. Les lignes (4, d) et (5, e) sont les lignes nouvellement insérées.

Écrire dans les tables Doris à l'aide d'une session Notebook

Copiez le code suivant dans l'onglet Notebook créé à l'étape précédente, mettez à jour les paramètres selon vos besoins, puis cliquez sur Run.

data = [(7, 'f'), (8, 'g')]
mockDataDF = spark.createDataFrame(data, ["id", "name"])
mockDataDF.write.mode("append").format("doris") \
  .option("doris.table.identifier", "testdb.test") \
  .option("doris.fenodes", "<doris_address>:<http_port>") \
  .option("user", "<user>") \
  .option("password", "<password>") \
  .save()
dorisSparkDF = spark.read.format("doris") \
  .option("doris.table.identifier", "testdb.test") \
  .option("doris.fenodes", "<doris_address>:<http_port>") \
  .option("user", "<user>") \
  .option("password", "<password>") \
  .load()
dorisSparkDF.show(10)

Si les données suivantes sont renvoyées, l'opération d'écriture a réussi.

+---+-----+
| id| name|
+---+-----+
|  1|    a|
|  2|    b|
|  3|    c|
|  7|    f|
|  8|    g|
+---+-----+