Dans cet atelier pratique, nous passons de la théorie à l'implémentation industrielle. Vous allez concevoir trois architectures de flux ETL complètes, répondant à des problématiques concrètes rencontrées quotidiennement en entreprise.

flowchart TD
    subgraph App1["Application 1 : Ingestion Multi-Fichiers"]
        A1["tFileList (Boucle sur les fichiers CSV)"] -->|Iterate| A2["tFileInputDelimited"]
        A2 --> A3["tMap & Insertion PostgreSQL"]
        A3 -->|OnSubjobOk| A4["Archivage automatique du fichier"]
    end
    subgraph App2["Application 2 : Jointure tMap & Rejets"]
        B1["Flux Principal : Transactions"] --> B3["tMap (Jointure Inner Join)"]
        B2["Flux Lookup : Table Clients"] --> B3
        B3 -->|Succès| B4["Entrepôt de Données"]
        B3 -->|Rejets| B5["Fichier d'Anomalies (Rejets)"]
    end
    subgraph App3["Application 3 : Synchronisation Inter-Bases"]
        C1["tPrejob : Ouverture des Sessions"] --> C2["Extraction Postgres -> tMap -> Upsert SQL Server"]
        C2 --> C3["tPostjob : Commit & Fermeture Sécurisée"]
    end
    style App1 fill:#e0f2fe,stroke:#0284c7,stroke-width:2px,color:#0f172a
    style App2 fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a
    style App3 fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a

---

Application 1 : Ingestion et archivage multi-fichiers en boucle

Problématique métier

Une chaîne de magasins dépose chaque nuit des fichiers CSV de tickets de caisse dans un dossier d'échange (C:/inbound/ventes_*.csv). Le nombre de fichiers varie selon l'activité. Le job doit parcourir le dossier, ingérer chaque fichier dans PostgreSQL, puis déplacer les fichiers traités dans un sous-dossier d'archive compressé.

Architecture des composants sur le canevas

tFileList_1 (Itération sur *.csv)
    |
    (Row > Iterate)
    v
tFileInputDelimited_1 ---> (Row > Main) ---> tMap_1 ---> (Row > Main) ---> tPostgresqlOutput_1
    |
    (Trigger > OnSubjobOk)
    v
tFileArchive_1 (Compression et archivage)

Configuration pas à pas

1. tFileList_1 :
2. Répertoire : "C:/inbound/"
3. Masque de fichier : "ventes_*.csv"
4. tFileInputDelimited_1 :
5. Nom de fichier : ((String)globalMap.get("tFileList_1_CURRENT_FILEPATH")) (Talend injecte dynamiquement le chemin du fichier courant de la boucle !)
6. Séparateur de champs : ";"
7. En-tête : 1
8. tMap_1 :
9. Normalise les montants et ajoute une colonne nom_fichier_source avec l'expression :

   ((String)globalMap.get("tFileList_1_CURRENT_FILE"))
   

10. tPostgresqlOutput_1 :
11. Action sur la table : Créer la table si elle n'existe pas.
12. Action sur les données : Insérer (Insert).
13. tFileArchive_1 :
14. Fichier archive : "C:/archive/archive_" + TalendDate.getDate("yyyyMMdd") + ".zip"
15. Cochez Supprimer les fichiers source après archivage.

---

Application 2 : Jointure relationnelle (Lookup) et gestion des rejets dans tMap

Problématique métier

Nous recevons un fichier de transactions financières ne contenant que l'identifiant du client (client_id) et le montant. Nous devons croiser ces transactions avec la table des clients dans PostgreSQL pour vérifier leur existence. Si le client n'existe pas en base de données, la ligne doit être déroutée vers un fichier CSV de rejets d'erreurs pour inspection par les auditeurs.

graph LR
    MAIN["tFileInputDelimited<br><i>Transactions (Main)</i>"] --> MAP["tMap"]
    LOOKUP["tPostgresqlInput<br><i>Clients (Lookup)</i>"] -->|Jointure sur client_id| MAP
    MAP -->|Flux Valide| DW["tPostgresqlOutput<br><i>ventes_consolidees</i>"]
    MAP -->|Inner Join Reject = true| REJ["tFileOutputDelimited<br><i>anomalies_clients_inconnus.csv</i>"]
    style MAIN fill:#f1f5f9,stroke:#64748b
    style LOOKUP fill:#e0f2fe,stroke:#0284c7
    style MAP fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a
    style DW fill:#dcfce7,stroke:#16a34a,color:#0f172a
    style REJ fill:#fee2e2,stroke:#ef4444,color:#0f172a

Configuration du tMap :

1. Reliez les transactions au tMap via Row > Main (row1).
2. Reliez le composant PostgreSQL au tMap via Row > Main (Talend le transforme automatiquement en flux Lookup row2).
3. Dans l'éditeur tMap :
4. Reliez la clé de jointure : glissez row1.client_id sur row2.id_client.
5. Cliquez sur l'icône de clé mécanique rouge de la table row2 :
   • Modèle de jointure : Jointure interne (Inner Join).
6. Créez deux tables de sortie :
7. sorties_valides : Récupère les colonnes de la transaction et le nom du client.
8. rejets_anomalies : Cliquez sur l'icône d'engrenage de cette table et réglez Rejet de jointure interne (Catch Output Reject) sur true.
9. Reliez sorties_valides à votre entrepôt PostgreSQL et rejets_anomalies à un composant tFileOutputDelimited.

[!NOTE] Grâce au mécanisme d'Inner Join Reject, pas une seule ligne corrompue ne vient polluer votre entrepôt décisionnel, et les données problématiques sont archivées pour audit.

---

Application 3 : Synchronisation inter-bases (PostgreSQL vers SQL Server) avec gestion transactionnelle

Problématique métier

Dans le cadre de la construction d'un Datamart sous Microsoft SQL Server, nous devons synchroniser quotidiennement les données d'une base opérationnelle sous PostgreSQL en utilisant le modèle transactionnel standard d'entreprise (tPrejob / tPostjob).

tPrejob_1 ---> (OnComponentOk) ---> tPostgresqlConnection_1 ---> (OnComponentOk) ---> tMSSqlConnection_1

tPostgresqlInput_1 ---> (Row > Main) ---> tMap_1 ---> (Row > Main) ---> tMSSqlOutput_1

tPostjob_1 ---> (OnComponentOk) ---> tMSSqlCommit_1 ---> (OnComponentOk) ---> tPostgresqlClose_1 ---> (OnComponentOk) ---> tMSSqlClose_1

Les règles d'or de la gestion transactionnelle :

1. tPrejob :
2. Ce composant s'exécute avant tout le reste du job.
3. Il ouvre les connexions physiques aux deux bases (tPostgresqlConnection et tMSSqlConnection) avec l'option Auto-commit désactivée.
4. Le flux de synchronisation :
5. tPostgresqlInput extrait les enregistrements modifiés à J-1.
6. Le composant tMSSqlOutput est configuré sur :
   • Action sur les données : Insérer ou Mettre à jour (Insert or Update).
   • Cochez l'option Utiliser une connexion existante pour réutiliser la session ouverte par le tPrejob.
7. tPostjob :
8. Ce composant s'exécute toujours à la toute fin du job, quoi qu'il arrive.
9. Si le flux a réussi sans accroc, tMSSqlCommit valide définitivement la transaction SQL.
10. tPostgresqlClose et tMSSqlClose ferment proprement les sockets réseau pour ne laisser aucune connexion fantôme sur les serveurs.

---

Checklist de qualité pour un Job Talend en Production

Avant de déclarer un flux d'intégration prêt pour l'exploitation industrielle :

• Les connexions utilisent les Métadonnées ou des variables de Contexte (aucun mot de passe codé en dur).
• Le job s'ouvre par un tPrejob et se clôture par un tPostjob avec fermeture systématique des connexions.
• Les gros volumes utilisent le mode Étendu (Extended Insert) ou un chunksize adapté dans les composants de sortie.
• La mémoire JVM est configurée dans le fichier .ini ou les paramètres d'exécution (-Xmx2048m).
• Les exceptions bloquantes sont capturées par tDie pour renvoyer un code de sortie non nul aux ordonnanceurs externes.

---

Vous maîtrisez désormais les architectures complètes d'intégration de données avec Talend. Dans la dernière leçon, nous passons en revue les ressources pour continuer à vous perfectionner et le quiz final d'auto-évaluation !

Partagez sur les réseaux sociaux

Commentaires