Exploit for data_pipeline_poc

Name: Exploit for data_pipeline_poc
Rating: 5 (229 reviews)
2025-09-08 | CVSS 8.2
## https://sploitus.com/exploit?id=947D1B5E-D22B-5348-B95A-F94973DA4080
# 📦 Catalogue Produit Unifié — Data Pipeline

## 🎯 Objectif

Ce projet est un **Proof of Concept (PoC)** visant à créer un pipeline automatisé d’intégration, transformation et orchestration de données produit issues de **trois systèmes hétérogènes** :

- **Infor M3**
- **Sage 100**
- **Akeneo**

L’objectif final est d’alimenter un **catalogue produit unifié** (`mart_articles`) dans Snowflake, exploitable par :

- des outils de data visualisation (ex. : **QlikView**)
- des **API B2B/B2C**
- avec une **mise à jour toutes les 15 minutes** (quasi temps réel).

---

## 🧱 Architecture Technique

```mermaid
flowchart TD
    subgraph Extraction [**Extraction - Talend**]
        A1[**Infor M3**M3CS_MITMAS,M3CS_CIDMAS,M3CS_OCUSMA] --> B1[m3_extract.bat]
        A2[**Sage 100**F_ARTICLE,F_COMPTET] --> B2[sage_extract.bat]
        A3[**Akeneo**PRODUCTS] --> B3[akeneo_extract.bat]
    end

    subgraph RawZone [**Snowflake - MIG.MPS**]
        B1 --> C1[(Raw Tables)]
        B2 --> C1
        B3 --> C1
    end

    subgraph Transformation [**Transformation - dbt**]
        C1 --> D1[(**STAGING**stg_m3_articles.sqlstg_sage_articles.sqlstg_akeneo_articles.sql)]
        D1 --> D2[(**INTERMEDIATE**int_articles.sql)]
        D2 --> D3[(**MART**mart_articles.sql)]
    end

    subgraph Orchestration [**Orchestration - Airflow**]
        F1[airflow_poc.py] --> B1
        F1 --> B2
        F1 --> B3
        F1 --> D1
        F1 --> D2
        F1 --> D3
    end

    D3 --> G1[**QlikView**]
    D3 --> G2[API pour consommation]
```
---

## 🧩 Structure du Projet

```text
data-pipeline-poc/
├── airflow/                 # Configuration Airflow + orchestration
│   ├── config/              # airflow.cfg ou custom .json
│   ├── dags/                # Tous les DAGs Airflow
│   │  	└── dag_talend_dbt_suppliers.py
│   ├── logs/                # Logs Airflow (volume)
│   ├── plugins/             # Plugins/hook Airflow
│   ├── .env                 # Variables d'environnement
│   └── docker-compose.yaml  # Docker orchestration (webserver, scheduler, etc.)
├── config/                  # Fichiers globaux de configuration
│   ├── connections.json     # Airflow connections à importer
│   └── variables.json       # Airflow variables à importer
│
├── dbt/                     # Projet dbt (dbt init ...)
│   ├── analyses/
│   ├── cert/
│   ├── docs/
│   ├── logs/
│   ├── macros/
│   ├── models/
│   ├── seeds/
│   ├── snapshots/
│   ├── target/
│   ├── tests/
│   ├── dbt_project.yml
│   ├── README.md
│   └── profiles.yml         # (optionnel : via ~/.dbt ou monté dans container)
│
├── docker/                  # Fichiers Docker supplémentaires
│   ├── airflow/             # Dockerfile custom si nécessaire
│   └── dbt/                 # Dockerfile dbt-only (si tu dockerises dbt)
│
├── scripts/                 # Scripts bash / python utilitaires
│   ├── init_airflow.sh
│   └── run_talend_job.sh
│
├── sql/                     # Scripts SQL manuels ou tests
│   └── init_snowflake.sql
│
├── talend_jobs/             # Export ou .bat/.sh des jobs Talend
│   └── m3_extract/
│   	├── POC_100_M3_API_LstSuppliers_0.1/
│       │    └── POC_100_M3_API_LstSuppliers/
│       │        └── POC_100_M3_API_LstSuppliers_run.sh
│   	└── POC_100_SNOWFLAKE_CIDMAS_0.1/
│            └── POC_100_SNOWFLAKE_CIDMAS/
│                └── POC_100_SNOWFLAKE_CIDMAS_run.sh
│
├── README.md
├── requirements.txt         # dépendances Python globales
└── .gitignore
```
### Explication de chaque dossier

| Dossier      | Rôle                                                                     |
|--------------|--------------------------------------------------------------------------|
| airflow/     | Tout ce qui concerne Airflow : orchestration, configuration, UI          |
| talend_jobs/ | Scripts ou exports des jobs Talend (exécutables .bat, .sh ou .jar)       |
| dbt/         | Projet dbt complet (transformation SQL dans Snowflake)                   |
| docker/      | Dockerfiles personnalisés si besoin (par ex. pour dbt ou airflow)        |
| sql/         | Fichiers SQL d'initialisation, tests ou débogage                         |
| config/      | Connexions, variables ou configs réutilisables (JSON, YAML)              |
| scripts/     | Utilitaires Bash ou Python pour automatiser certaines étapes             |
| .env         | Variables globales (Snowflake creds, chemin Talend, etc.)                |


### Cycle de données :
- **Talend :** extraction depuis M3, Sage ou Akeneo → fichiers ou DB
- **Airflow :** planification et exécution des .bat Talend ou des .sh
- **Snowflake :** Stockage des données venant de Talend → schéma STAGING
- **dbt (appelé par Airflow) :** transformation des données → MART

Résultats consultables dans **QlikView** ou exposés par **API**.

**Objectifs :**
---

## 🔄 Fonctionnement du Pipeline

### 1. 🔹 Extraction (Talend)

Des scripts batch déclenchent des jobs Talend pour extraire les données des systèmes sources :

| Source   | Script                  | Tables extraites                                 |
|----------|-------------------------|--------------------------------------------------|
| Infor M3 | `m3_extract_run.bat`    | `M3CS_MITMAS`, `M3CS_CIDMAS`, `M3CS_OCUSMA`      |
| Sage 100 | `sage_extract_run.bat`  | `F_ARTICLE`, `F_COMPTET`                         |
| Akeneo   | `akeneo_extract_run.bat`| `PRODUCTS`                                       |

Les données sont chargées dans **Snowflake**, schéma **`MIG.MPS`** (zone RAW).

---

### 2. 🧪 Transformation (dbt)

Les modèles dbt effectuent une transformation SQL en trois couches :

**Objectifs :**
Nettoyer, standardiser, enrichir et unifier les données via modèles SQL dbt répartis dans 3 couches :

#### STAGING : Nettoyage initial (Bronze)
**Nom des modèles :**
- stg_m3_articles.sql
- stg_m3_fournisseurs.sql
- stg_m3_clients.sql
- stg_sage_articles.sql
- stg_sage_fournisseurs.sql
- stg_sage_clients.sql
- stg_akeneo_articles.sql

**Opérations typiques :**
- Nettoyage des colonnes (trim, upper, rename, formats).
- Cast des types de données.
- Alignement des formats de clé produit (ex : références, code articles, SKU...).
- Ajout de métadonnées (date chargement, source...).

#### INTERMEDIATE : Fusion et enrichissement (Silver)
**Modèle :**
- int_articles.sql
- int_fournisseurs.sql
- int_clients.sql

**Objectifs :**
- Fusion logique entre les différentes sources (match sur clé produit).
- Jointures avec les fournisseurs et clients si pertinent.
- Calculs intermédiaires (statuts, prix consolidés, sources dominantes...).

#### MART : Modèle final exposé (Gold)
**Modèle :**
- mart_articles.sql
- mart_fournisseurs.sql
- mart_clients.sql

**Objectifs :**
- Vue unifiée du catalogue produit.
- Prête à consommation par les outils QlikView et les API.
- Table matérialisée ou vue selon performance/consistance.

#### Fichiers clés :
- `sources.yml` : déclaration des tables RAW
- `profiles.yml` : configuration dbt vers Snowflake via clé PEM

---

### 3. 🛠️ Orchestration (Apache Airflow)

Le DAG `airflow_poc.py` orchestre tout le pipeline toutes les **15 minutes** :

```text
extract_m3 → extract_sage → extract_akeneo → dbt_run → dbt_test
```

## ⚙️ Technologies Utilisées

| Composant        | Version     | Rôle principal                                   |
|------------------|-------------|--------------------------------------------------|
| Python           | 3.11.4      | Langage principal pour les scripts et Airflow    |
| Talend Open Studio | 8.x       | Extraction des données (ETL)                     |
| dbt Core         | 1.x         | Transformation des données (SQL Modelling)       |
| Apache Airflow   | 2.8.4       | Orchestration des workflows                      |
| Snowflake        | Cloud       | Data Warehouse cible                             |
| Git / GitHub     | Dernière    | Versionnement du code                            |
| Windows          | 10/11       | Environnement d’exécution                        |

## ✅ Contrôles & Tests

### Tests dbt
- `dbt test` est exécuté automatiquement après chaque `dbt run`.
- Tests inclus :
  - Vérification de colonnes non nulles (`not_null`)
  - Unicité des identifiants (`unique`)
  - Intégrité des jointures entre sources

### Validation des extractions Talend
- Log Talend vérifié à chaque exécution.
- Retour d’état capturé dans Airflow.

### Monitoring de base
- Airflow UI permet le suivi manuel de l'état de chaque tâche.
- Statut de réussite/échec visible sur les DAGs.

> Pour un déploiement en production, prévoir l’ajout de tests de qualité via **Great Expectations** ou **dbt tests personnalisés**.

## 🔐 Sécurité & Accès

### Connexion à Snowflake
- Authentification via clé PEM (stockée localement de manière sécurisée)
- Mot de passe de la clé privé référencé par la variable Airflow `SNOWFLAKE_KEYSTORE_PASS`

### Recommandations
- Ne pas versionner la clé privée ou `profiles.yml` contenant les identifiants.
- Utiliser `.gitignore` pour exclure tout fichier sensible.
- Séparer les environnements dev/prod via des profils dbt distincts.

## 🚀 Améliorations Possibles

- 🔁 Passage d’Airflow vers un exécuteur parallèle (`LocalExecutor` ou `CeleryExecutor`) pour production.
- 🧪 Intégration de **tests de qualité des données** avec [Great Expectations](https://greatexpectations.io/).
- 📊 Monitoring étendu avec **Prometheus + Grafana** ou **Airflow + DataDog**.
- 🔒 Utilisation de **Vault** pour la gestion des secrets.
- 📦 Conteneurisation via **Docker** et déploiement dans **AWS ECS / EKS**.
- 🧱 Refonte du modèle `mart_articles` selon les principes de **modélisation en étoile**.
- 📁 Implémentation de CI/CD avec **GitHub Actions** ou **GitLab CI** pour automatiser tests et déploiements.