Dans le cadre de la gestion de courses de VTC, nous faisions face à une problématique de requête complexe. Le système devait interroger six types d'événements temporels distincts pour chaque commande : l'heure de diffusion, l'heure de prise en charge efficace, l'heure de confirmation, l'heure d'annulation, l'heure d'arrivée et l'heure de paiement. Ces données provenaient de sources variées comme MongoDB et des flux Kafka en temps réel.

En raison du comportement des passagers qui ne paient pas toujours immédiatement après la course, une jointure directe entre les tables était impossible sans fausser les résultats. La solution adoptée consistait à requêter une table unique regroupant tous ces événements, puis à effectuer une opération d'union pour combiner les résultats avant de les stocker dans HBase, accessible via Apache Phoenix.

La requête initiale exécutée sur Phoenix ressemblait à ceci :

SELECT
    T1.id_chauffeur,
    T1.nom_chauffeur,
    T1.telephone,
    T1.nom_alliance,
    T1.ville_inscription,
    T1.type_chauffeur,
    SUM(T1.nb_diffusions) AS "nb_diffusions",
    SUM(T1.nb_prises_efficaces) AS "nb_prises_efficaces",
    SUM(T1.nb_confirmations) AS "nb_confirmations",
    SUM(T1.nb_annulations) AS "nb_annulations",
    SUM(T1.nb_annulations_passagers) AS "nb_annulations_passagers",
    SUM(T1.nb_annulations_chauffeurs) AS "nb_annulations_chauffeurs",
    SUM(T1.nb_annulations_service) AS "nb_annulations_service",
    SUM(T1.nb_arrivees) AS "nb_arrivees",
    SUM(T1.nb_paiements) AS "nb_paiements",
    SUM(T1.nb_cumul_commandes) AS "nb_cumul_commandes",
    SUM(T1.nb_cumul_paiements) AS "nb_cumul_paiements"
FROM (
    -- Premier bloc pour les diffusions
    SELECT
        "id_chauffeur",
        "nom_chauffeur",
        "telephone",
        "nom_alliance",
        "ville_inscription",
        "type_chauffeur",
        SUM(TO_NUMBER("nb_diffusions")) AS "nb_diffusions",
        0 AS "nb_prises_efficaces",
        0 AS "nb_confirmations",
        0 AS "nb_annulations",
        0 AS "nb_annulations_passagers",
        0 AS "nb_annulations_chauffeurs",
        0 AS "nb_annulations_service",
        0 AS "nb_arrivees",
        0 AS "nb_paiements",
        0 AS "nb_cumul_commandes",
        0 AS "nb_cumul_paiements"
    FROM "reservation_globale"
    WHERE
        CASE WHEN "heure_diffusion" != '0' THEN
            TO_DATE("heure_diffusion",'yyyy-MM-dd') >= TO_DATE(#{debut},'yyyy-MM-dd')
            AND TO_DATE("heure_diffusion",'yyyy-MM-dd') <= TO_DATE(#{fin},'yyyy-MM-dd')
        ELSE 1=1 END
    GROUP BY "id_chauffeur", "nom_chauffeur", "telephone", "nom_alliance", "ville_inscription", "type_chauffeur"

    UNION ALL

    -- Deuxième bloc pour les prises efficaces
    SELECT
        "id_chauffeur",
        "nom_chauffeur",
        "telephone",
        "nom_alliance",
        "ville_inscription",
        "type_chauffeur",
        0 AS "nb_diffusions",
        SUM(TO_NUMBER("nb_prises_efficaces")) AS "nb_prises_efficaces",
        0 AS "nb_confirmations",
        0 AS "nb_annulations",
        0 AS "nb_annulations_passagers",
        0 AS "nb_annulations_chauffeurs",
        0 AS "nb_annulations_service",
        0 AS "nb_arrivees",
        0 AS "nb_paiements",
        0 AS "nb_cumul_commandes",
        0 AS "nb_cumul_paiements"
    FROM "reservation_globale"
    WHERE
        CASE WHEN "heure_prise_efficace" != '0' THEN
            TO_DATE("heure_prise_efficace",'yyyy-MM-dd') >= TO_DATE(#{debut},'yyyy-MM-dd')
            AND TO_DATE("heure_prise_efficace",'yyyy-MM-dd') <= TO_DATE(#{fin},'yyyy-MM-dd')
        ELSE 1=1 END
    GROUP BY "id_chauffeur", "nom_chauffeur", "telephone", "nom_alliance", "ville_inscription", "type_chauffeur"

    -- ... blocs similaires pour les autres événements ...
) AS T1
GROUP BY T1.id_chauffeur, T1.nom_chauffeur, T1.telephone, T1.nom_alliance, T1.ville_inscription, T1.type_chauffeur;

Avec environ deux millions d'enregistrements, cette requête sans index sur Phoenix prenait plus de six secondes. Chaque sous-requête dans l'union nécessitait environ une seconde de scan complet, et l'agrégation finale ajoutait un surcoût.

L'analyse a révélé que le problème principle venait du volume de données et des scans répétés. Pour y remédier, nous avons implémenté une stratégie de pré-traitement des données avec Apache Spark. L'objectif était de "dégraisser" les données en les consolidant par jour et par chauffeur, transformant les horodatages précis en dates au format 'yyyy-MM-dd'.

Le processus Spark commençait par charger la table principale et la mettre en cache :

val donneesReservation = sparkSession.read
    .format("org.apache.spark.sql.execution.datasources.hbase")
    .options(Map(
        "catalog" -> catalogueHBase,
        "table" -> "reservation_source"
    )).load()

donneesReservation.createOrReplaceTempView("vue_reservation")
sparkSession.sqlContext.cacheTable("vue_reservation")

Ensuite, nous avons créé six DataFrames pré-agrégés, chacun correspondant à un type d'événement. Voici un exemple pour les diffusions :

val dfDiffusions = sparkSession.sql("""
    SELECT
        chauffeur_id,
        nom,
        mobile,
        alliance,
        gestionnaire_id,
        ville_inscription,
        type_vehicule,
        DATE_FORMAT(moment_diffusion, 'yyyy-MM-dd') AS date_evenement,
        COUNT(*) AS compteur
    FROM vue_reservation
    WHERE moment_diffusion IS NOT NULL
        AND moment_diffusion != '0'
        AND DATE_FORMAT(moment_diffusion, 'yyyyMMdd') < DATE_FORMAT(CURRENT_DATE(), 'yyyyMMdd')
    GROUP BY chauffeur_id, nom, mobile, alliance, gestionnaire_id, ville_inscription, type_vehicule, date_evenement
""")

Ces DataFrames étaient ensuite persistés dans HBase via Spark, créant ainsi des tables dénormalisées et allégées. Chaque table contenait les agrégations quotidiennes pour un événement spécifique.

La requête Phoenix finale a été modifiée pour interroger ces six nouvelles tables. Bien que la structure d'union soit préservée, le volume de données scannées par requête a drastiquement diminué :

SELECT
    TA.id_chauffeur,
    TA.nom,
    TA.mobile,
    TA.alliance,
    TA.ville,
    TA.type_vehicule,
    SUM(TA.compteur_diff) AS "total_diffusions",
    SUM(TA.compteur_prise) AS "total_prises_efficaces",
    SUM(TA.compteur_confirm) AS "total_confirmations",
    SUM(TA.compteur_annul) AS "total_annulations",
    SUM(TA.compteur_annul_pass) AS "total_annulations_passagers",
    SUM(TA.compteur_annul_chauff) AS "total_annulations_chauffeurs",
    SUM(TA.compteur_annul_serv) AS "total_annulations_service",
    SUM(TA.compteur_arrivee) AS "total_arrivees",
    SUM(TA.compteur_paiement) AS "total_paiements",
    SUM(TA.compteur_cumul_cmd) AS "cumul_commandes",
    SUM(TA.compteur_cumul_pay) AS "cumul_paiements"
FROM (
    SELECT
        chauffeur_id AS "id_chauffeur",
        nom_complet AS "nom",
        numero_mobile AS "mobile",
        nom_alliance AS "alliance",
        ville_enregistrement AS "ville",
        categorie_chauffeur AS "type_vehicule",
        compteur AS "compteur_diff",
        0 AS "compteur_prise",
        0 AS "compteur_confirm",
        0 AS "compteur_annul",
        0 AS "compteur_annul_pass",
        0 AS "compteur_annul_chauff",
        0 AS "compteur_annul_serv",
        0 AS "compteur_arrivee",
        0 AS "compteur_paiement",
        0 AS "compteur_cumul_cmd",
        0 AS "compteur_cumul_pay"
    FROM "tb_diffusions_agregees"
    WHERE date_evenement BETWEEN TO_DATE(#{debut},'yyyy-MM-dd') AND TO_DATE(#{fin},'yyyy-MM-dd')

    UNION ALL

    SELECT
        chauffeur_id AS "id_chauffeur",
        nom_complet AS "nom",
        numero_mobile AS "mobile",
        nom_alliance AS "alliance",
        ville_enregistrement AS "ville",
        categorie_chauffeur AS "type_vehicule",
        0 AS "compteur_diff",
        compteur AS "compteur_prise",
        0 AS "compteur_confirm",
        0 AS "compteur_annul",
        0 AS "compteur_annul_pass",
        0 AS "compteur_annul_chauff",
        0 AS "compteur_annul_serv",
        0 AS "compteur_arrivee",
        0 AS "compteur_paiement",
        0 AS "compteur_cumul_cmd",
        0 AS "compteur_cumul_pay"
    FROM "tb_prises_efficaces_agregees"
    WHERE date_evenement BETWEEN TO_DATE(#{debut},'yyyy-MM-dd') AND TO_DATE(#{fin},'yyyy-MM-dd')

    -- ... unions pour les quatre autres tables agrégées ...
) AS TA
GROUP BY TA.id_chauffeur, TA.nom, TA.mobile, TA.alliance, TA.ville, TA.type_vehicuel

Cette approche de pré-agrégation a permis de réduire le temps d'exécution de la requête de plus de six secondes à moins d'une seconde, même sans ajouter d'index sur les tables Phoenix, grâce à la réduction drastique du volume de données à scanner lors de l'exécution.