Jusqu'à présent, nous avons exécuté des blocs anonymes (DO $$ ... $$). Bien qu'utiles pour des tests ponctuels, les blocs anonymes ne peuvent ni recevoir de paramètres dynamiques de l'extérieur, ni être réutilisés par d'autres applications.
Pour bâtir un système d'information maintenable et performant, les Data Engineers encapsulent leur logique dans des Fonctions (User-Defined Functions) et des Procédures Stockées (Stored Procedures).
1. Pourquoi encapsuler du code dans le serveur SQL ?#
- Performances accrues : Le code PL/pgSQL est analysé, compilé sous forme d'arbre d'exécution et mis en cache mémoire sur le serveur.
- Réduction massive du trafic réseau : Au lieu de transférer 500 000 lignes brutes vers un script Python pour faire un traitement pour ensuite les renvoyer en base, tout s'exécute directement en local sur le moteur SQL.
- Sécurité et gouvernance : Vous pouvez interdire l'accès direct en écriture aux tables sensibles aux utilisateurs et leur accorder uniquement le droit
GRANT EXECUTEsur une procédure validée.
2. Procédure vs Fonction : Quelle est la différence ?#
C'est une des questions les plus posées en entretien technique Data :
| Critère | Fonction (FUNCTION) |
Procédure Stockée (PROCEDURE) |
|---|---|---|
| Valeur de retour | Obligatoire (RETURNS type ou RETURNS TABLE) |
Aucune (ou via paramètres OUT) |
| Commande d'appel | Utilisée dans un SELECT : SELECT ma_fonction(42); |
Invoquée via CALL : CALL ma_procedure('2024-03-01'); |
| Contrôle transactionnel | Impossible : fait partie de la transaction appelante | Autorisé : peut faire COMMIT; ou ROLLBACK; |
| Usage privilégié | Calculs mathématiques, transformations, vues paramétrées | Pipelines ETL, purges d'archivage, maintenance par lots |
[!IMPORTANT] La différence capitale : Depuis PostgreSQL 11, seules les procédures stockées (
PROCEDURE) peuvent exécuter des instructions de validation ou d'annulation de transactions (COMMIT;etROLLBACK;). Si vous devez traiter 10 millions de lignes par blocs de 100 000 avec un commit régulier, vous devez impérativement utiliser une procédure !
3. Les Fonctions scalaires (CREATE FUNCTION)#
Une fonction scalaire reçoit des arguments et renvoie une valeur unique.
Exemple : Calcul d'un prix remisé#
CREATE OR REPLACE FUNCTION analytics.fn_calculer_prix_remise(
p_prix_ht NUMERIC,
p_taux_remise NUMERIC DEFAULT 0.05
)
RETURNS NUMERIC
LANGUAGE plpgsql
AS
$$
DECLARE
v_prix_final NUMERIC;
BEGIN
-- Contrôle d'intégrité sur les arguments
IF p_prix_ht < 0 THEN
RAISE EXCEPTION 'Le prix unitaire ne peut pas être négatif : %', p_prix_ht;
END IF;
v_prix_final := p_prix_ht * (1 - p_taux_remise);
RETURN ROUND(v_prix_final, 2);
END;
$$;
Appel dans une requête SQL standard :#
SELECT
nom_produit,
prix_catalogue,
analytics.fn_calculer_prix_remise(prix_catalogue, 0.15) AS prix_solde
FROM prod.produits;
4. Les Fonctions retournant des tables (RETURNS TABLE)#
Une fonction peut également retourner un jeu de données complet. Elle se comporte alors comme une vue dynamique paramétrée :
CREATE OR REPLACE FUNCTION reporting.fn_top_clients_par_region(
p_region VARCHAR,
p_limite INTEGER DEFAULT 5
)
RETURNS TABLE (
client_id INTEGER,
nom_client VARCHAR,
chiffre_affaires NUMERIC
)
LANGUAGE plpgsql
AS
$$
BEGIN
RETURN QUERY
SELECT
c.id_client,
c.nom,
SUM(v.montant) AS total_ca
FROM analytics.dim_clients c
JOIN analytics.fct_ventes v ON c.id_client = v.id_client
WHERE c.region = p_region
GROUP BY c.id_client, c.nom
ORDER BY total_ca DESC
LIMIT p_limite;
END;
$$;
Interrogation comme une table native :#
-- Récupération du Top 3 des clients de la région 'Île-de-France'
SELECT * FROM reporting.fn_top_clients_par_region('Île-de-France', 3);
5. Les Procédures Stockées avec transactions (CREATE PROCEDURE)#
Les procédures sont créées avec le mot-clé CREATE OR REPLACE PROCEDURE et invoquées avec l'instruction CALL.
CREATE OR REPLACE PROCEDURE dwh.sp_archiver_donnees_anciennes(
p_date_limite DATE,
p_taille_lot INTEGER DEFAULT 50000
)
LANGUAGE plpgsql
AS
$$
DECLARE
v_lignes_supprimees INTEGER := 0;
BEGIN
RAISE NOTICE 'Démarrage de l archivage des données antérieures au %', p_date_limite;
LOOP
-- 1. Copie d'un lot vers la table d'archives historiques
WITH lot_a_deplacer AS (
DELETE FROM prod.logs_activite
WHERE ctid IN (
SELECT ctid
FROM prod.logs_activite
WHERE date_creation < p_date_limite
LIMIT p_taille_lot
)
RETURNING *
)
INSERT INTO archives.logs_activite_histo
SELECT * FROM lot_a_deplacer;
GET DIAGNOSTICS v_lignes_supprimees = ROW_COUNT;
RAISE NOTICE 'Lot de % enregistrements archivé et purgé.', v_lignes_supprimees;
-- 2. Validation immédiate de la transaction pour libérer le verrou
COMMIT;
-- 3. Sortie dès que le dernier lot contient moins que la taille cible
EXIT WHEN v_lignes_supprimees < p_taille_lot;
END LOOP;
RAISE NOTICE 'Opération d archivage terminée avec succès.';
END;
$$;
Exécution de la procédure :#
6. Architecture d'un Pipeline ETL complet en PL/pgSQL#
Voici comment orchestrer un pipeline ETL de production avec gestion d'erreurs intégrée :
CREATE OR REPLACE PROCEDURE etl.sp_charger_datamart_ventes(p_date_run DATE)
LANGUAGE plpgsql
AS
$$
DECLARE
v_start_time TIMESTAMP := clock_timestamp();
v_count_ins INTEGER := 0;
BEGIN
RAISE NOTICE '>>> Lancement ETL Datamart Ventes pour la date: %', p_date_run;
-- ÉTAPE 1 : Purge préalable de la partition cible (idempotence)
DELETE FROM datamart.fct_ventes_synthese WHERE date_vente = p_date_run;
-- ÉTAPE 2 : Transformation et insertion
INSERT INTO datamart.fct_ventes_synthese (date_vente, id_produit, total_quantite, total_ca)
SELECT
date_commande,
id_produit,
SUM(quantite),
SUM(quantite * prix_unitaire)
FROM staging.commandes
WHERE date_commande = p_date_run
GROUP BY date_commande, id_produit;
GET DIAGNOSTICS v_count_ins = ROW_COUNT;
-- ÉTAPE 3 : Journalisation dans la table de log
INSERT INTO etl.pipeline_logs (nom_job, date_cible, lignes_traitees, duree_ms, statut)
VALUES ('sp_charger_datamart_ventes', p_date_run, v_count_ins,
EXTRACT(MILLISECONDS FROM clock_timestamp() - v_start_time), 'SUCCESS');
COMMIT;
RAISE NOTICE '>>> ETL achevé avec succès : % lignes insérées en % ms',
v_count_ins, EXTRACT(MILLISECONDS FROM clock_timestamp() - v_start_time);
EXCEPTION
WHEN OTHERS THEN
ROLLBACK;
INSERT INTO etl.pipeline_logs (nom_job, date_cible, message_erreur, statut)
VALUES ('sp_charger_datamart_ventes', p_date_run, SQLERRM, 'FAILED');
COMMIT;
RAISE EXCEPTION 'Échec critique du pipeline ETL : %', SQLERRM;
END;
$$;
7. Mise en pratique : Créer une fonction de scoring client#
Énoncé#
Créez une fonction fn_score_client(p_client_id INTEGER) RETURNS VARCHAR qui :
1. Calcule le montant total des dépenses du client dans fct_ventes.
2. Si le client n'a jamais commandé, renvoie 'NOUVEAU'.
3. S'il a dépensé moins de 1 000 €, renvoie 'BRONZE'.
4. Entre 1 000 € et 5 000 €, renvoie 'ARGENT'.
5. Au-delà de 5 000 €, renvoie 'OR'.
CREATE OR REPLACE FUNCTION analytics.fn_score_client(p_client_id INTEGER)
RETURNS VARCHAR
LANGUAGE plpgsql
AS
$$
DECLARE
v_total_depenses NUMERIC;
BEGIN
SELECT COALESCE(SUM(montant), 0)
INTO v_total_depenses
FROM analytics.fct_ventes
WHERE id_client = p_client_id;
IF v_total_depenses = 0 THEN
RETURN 'NOUVEAU';
ELSIF v_total_depenses < 1000 THEN
RETURN 'BRONZE';
ELSIF v_total_depenses <= 5000 THEN
RETURN 'ARGENT';
ELSE
RETURN 'OR';
END IF;
END;
$$;
Félicitations ! Vous maîtrisez désormais l'ensemble de la programmation procédurale avec SQL et PL/pgSQL : variables, structures conditionnelles, itérateurs et sous-programmes modulaires transactionnels.
Commentaires