E-MapReduce:Développement SparkSQL

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Un espace de travail. Consultez la rubrique Créer un espace de travail.

• Une instance de session SQL. Consultez la rubrique Gérer les sessions SQL.

Créer une tâche Spark SQL

1. Accédez à la page de développement.

   1. Connectez-vous à la console EMR.

   2. Dans le volet de navigation de gauche, choisissez EMR Serverless \> Spark.

   3. Sur la page Spark, cliquez sur le nom de l'espace de travail cible.

   4. Sur la page EMR Serverless Spark, cliquez sur Development dans le volet de navigation de gauche.

2. Créez une tâche.

   1. Sous l'onglet Development, cliquez sur l'icône .

   2. Dans la boîte de dialogue, saisissez un Name, définissez le type sur SparkSQL et cliquez sur OK.

   3. Dans le coin supérieur droit, sélectionnez un catalogue de données, une base de données et une instance de session SQL en cours d'exécution. Pour créer une nouvelle instance de session SQL, sélectionnez Connect to SQL Session dans la liste déroulante. Consultez la rubrique Gérer les sessions SQL pour plus de détails.

   4. Saisissez les instructions SQL dans l'éditeur.

      Exemple 1 : Opérations SQL de base

      Créez une base de données, basculez vers celle-ci, créez une table, insérez des lignes et interrogez les données.

      create DATABASE test_sql;
      
      use test_sql;
      
      CREATE TABLE IF NOT EXISTS my_table (id INT, name STRING);
      
      INSERT INTO my_table VALUES(1, 'Alice'),(2, 'Bob');
      
      SELECT * FROM my_table WHERE id > 1;

      Les résultats s'affichent dans le panneau de sortie sous l'éditeur.

      #	id	name
      1	2	Bob

      Exemple 2 : Table externe basée sur CSV

      Créez une table externe adossée à un fichier CSV dans Object Storage Service (OSS) et exécutez une requête analytique. Remplacez oss://<bucketname>/user/ par le chemin réel de votre bucket.

      1. Créez la table externe. Définissez une table nommée orders avec les champs suivants :

         • order_id : ID de la commande.

         • order_date : Horodatage de la commande, par exemple '2025-07-01 10:00:00'.

         • order_category : Catégorie du produit, par exemple 'Electronics' ou 'Apparel'.

         • order_revenue : Montant de la commande.

         CREATE TABLE orders (
           order_id       STRING,   -- Order ID
           order_date     STRING,   -- Order timestamp
           order_category STRING,   -- Product category
           order_revenue  DOUBLE    -- Order amount
         )
         USING CSV
         OPTIONS (
           path   'oss://<bucketname>/user/',
           header 'true'
         );

      2. Insérez des données de test.

         INSERT OVERWRITE TABLE orders VALUES
           ('o1',  '2025-07-01 10:00:00', 'Electronics', 5999.0),
           ('o2',  '2025-07-02 11:30:00', 'Apparel',     299.0),
           ('o3',  '2025-07-03 14:45:00', 'Electronics', 899.0),
           ('o4',  '2025-07-04 09:15:00', 'Home Goods',  99.0),
           ('o5',  '2025-07-05 16:20:00', 'Electronics', 1999.0),
           ('o6',  '2025-07-06 08:00:00', 'Apparel',     199.0),
           ('o7',  '2025-07-07 12:10:00', 'Electronics', 799.0),
           ('o8',  '2025-07-08 18:30:00', 'Home Goods',  59.0),
           ('o9',  '2025-07-09 20:00:00', 'Electronics', 399.0),
           ('o10', '2025-07-10 07:45:00', 'Apparel',     599.0),
           ('o11', '2025-07-11 09:00:00', 'Electronics', 1299.0),
           ('o12', '2025-07-12 13:20:00', 'Home Goods',  159.0),
           ('o13', '2025-07-13 17:15:00', 'Apparel',     499.0),
           ('o14', '2025-07-14 21:30:00', 'Electronics', 999.0),
           ('o15', '2025-07-15 06:10:00', 'Home Goods',  299.0);

      3. Exécutez une requête analytique. La requête suivante renvoie les performances de vente par catégorie sur une période de 15 jours — nombre de commandes, valeur marchande brute totale (GMV), montant moyen de la commande et date de la dernière commande — pour les catégories dont le revenu total est supérieur à 1 000, triées par GMV décroissant.

         SELECT
             order_category,
             COUNT(order_id)    AS order_count,
             SUM(order_revenue) AS gmv,
             AVG(order_revenue) AS avg_order_amount,
             MAX(order_date)    AS latest_order_date
         FROM orders
         WHERE
             CAST(order_date AS TIMESTAMP) BETWEEN '2025-07-01' AND '2025-07-15'
         GROUP BY order_category
         HAVING SUM(order_revenue) > 1000
         ORDER BY gmv DESC, order_category ASC;

      Exemple 3 : Créer une table Hive Parquet (table de format DLF)

      Par défaut, EMR Serverless Spark utilise Apache Paimon comme format de table. Pour créer une table Hive Parquet à la place, vous devez remplacer les paramètres par défaut dans les paramètres de démarrage de la tâche.

      Paramètres de démarrage

      Avant de créer une table Hive Parquet, ajoutez les paramètres de configuration Spark suivants pour empêcher le catalogue Paimon d'intercepter la création de la table :

      spark.sql.catalog.spark_catalog=org.apache.spark.sql.hive.HiveCatalog
      spark.sql.defaultFileFormat=parquet

      Vous pouvez ajouter ces paramètres dans la section Configuration de la console EMR Serverless Spark ou en utilisant le paramètre --conf dans la CLI.

      Méthode SQL

      Utilisez la syntaxe CREATE TABLE ... USING PARQUET pour créer une table Hive Parquet. N'utilisez pas STORED AS PARQUET.

      CREATE TABLE my_parquet_table (
        id INT,
        name STRING,
        created_at TIMESTAMP
      ) USING PARQUET;

      Méthode API DataFrame

      Utilisez df.write.format("parquet").saveAsTable(...) pour écrire les données sous forme de table Hive Parquet.

      df.write.format("parquet").saveAsTable("my_parquet_table")

      Remarque

      Si votre espace de travail est lié à un catalogue DLF, nous vous recommandons de créer des tables à l'aide de sessions Spark ou d'instructions SQL afin de garantir la cohérence avec le catalogue de métadonnées.

   5. (Facultatif) Cliquez sur l'onglet Version Information à droite pour comparer les versions. L'éditeur met en évidence les différences de code SQL entre les versions.

3. Exécutez et publiez la tâche.

   1. Cliquez sur Run. Les résultats apparaissent sous l'onglet Execution Results. En cas d'erreur, vérifiez l'onglet Execution Issues. Le panneau d'historique d'exécution à droite affiche les enregistrements des trois derniers jours.

   2. Après avoir confirmé que la tâche s'exécute correctement, cliquez sur Publish dans le coin supérieur droit.

   3. Dans la boîte de dialogue Publish, saisissez les notes de version et cliquez sur OK.

Accéder à l'interface utilisateur Spark (Spark UI)

L'interface utilisateur Spark (Spark UI) affiche l'état d'exécution des tâches, l'utilisation des ressources et les informations des journaux, ce qui s'avère utile pour analyser et optimiser vos tâches Spark.

Raccourcis clavier

Étapes suivantes

Planifiez l'exécution périodique de la tâche en créant un workflow. Consultez la rubrique Créer un workflow. Pour un exemple complet de planification, consultez la rubrique Démarrage rapide pour le développement Spark SQL.

FAQ