Chargement et transformation

Dockerfile

@@ -1,3 +1,10 @@
 # FROM astrocrpublic.azurecr.io/runtime:3.0-10
 FROM quay.io/astronomer/astro-runtime:11.4.0
 
+# Copy pip configuration for better network handling
+COPY pip.conf /etc/pip.conf
+
+# Set environment variables for pip
+ENV PIP_TIMEOUT=300
+ENV PIP_RETRIES=5
+

README.md

@@ -52,12 +52,28 @@ Ce projet a pour objectif de concevoir un **pipeline ELT complet** permettant d
 git clone https://github.com/votre-utilisateur/youtube-elt-pipeline.git
 cd youtube-elt-pipeline
 
+# Créer un fichier .env avec vos clés API
+cp .env.example .env
+# Éditer .env avec vos clés YouTube API
+
 # Démarrer le projet Airflow avec Astro
 astro dev start
 
+# Si le build échoue à cause de timeouts réseau, essayer:
+# 1. Attendre et relancer: astro dev start
+# 2. Ou installer les dépendances optionnelles après le démarrage:
+#    astro dev bash
+#    python /opt/airflow/scripts/install_optional_deps.py
+
 # Lancer les DAGs depuis l'UI Airflow
 ```
 
+### Résolution des problèmes de build
+Si vous rencontrez des timeouts lors de l'installation des dépendances Python:
+1. Le fichier `requirements.txt` contient uniquement les dépendances essentielles
+2. Les dépendances optionnelles (soda-core, streamlit, etc.) sont dans `requirements-optional.txt`
+3. Vous pouvez les installer après le démarrage du container avec le script fourni
+
 ### CI/CD
 - Workflow GitHub Actions : build Docker, tests unitaires et intégration, vérification DAGs.
 

dags/youtube_elt_dag.py

@@ -5,6 +5,7 @@
 import os
 from pathlib import Path
 import subprocess
+import sys
 
 DEFAULT_ARGS = {
     "owner": "etl",
@@ -20,10 +21,26 @@
          catchup=False) as dag:
 
     def run_extract():
-        subprocess.check_call(["python", "/opt/airflow/scripts/extract_youtube.py"])
+        try:
+            # Ensure data directory exists
+            os.makedirs("/opt/airflow/data", exist_ok=True)
+            subprocess.check_call(["python", "/opt/airflow/scripts/extract_youtube.py"])
+        except subprocess.CalledProcessError as e:
+            print(f"Extract failed with return code {e.returncode}")
+            sys.exit(1)
+        except Exception as e:
+            print(f"Extract failed with error: {str(e)}")
+            sys.exit(1)
 
     def run_load():
-        subprocess.check_call(["python", "/opt/airflow/scripts/load_to_postgres.py"])
+        try:
+            subprocess.check_call(["python", "/opt/airflow/scripts/load_to_postgres.py"])
+        except subprocess.CalledProcessError as e:
+            print(f"Load failed with return code {e.returncode}")
+            sys.exit(1)
+        except Exception as e:
+            print(f"Load failed with error: {str(e)}")
+            sys.exit(1)
 
     t1 = PythonOperator(task_id="extract", python_callable=run_extract)
     t2 = PythonOperator(task_id="load", python_callable=run_load)
@@ -34,9 +51,42 @@ def run_load():
     )
 
     def soda_check():
-        # Exécute soda scan (doit être installé). Exemple simple via subprocess.
-        subprocess.check_call(["soda", "scan", "--scan-yaml", "/opt/airflow/soda/checks.yml"])
+        try:
+            # Check if soda is available
+            subprocess.check_call(["which", "soda"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            # Exécute soda scan (doit être installé). Exemple simple via subprocess.
+            subprocess.check_call(["soda", "scan", "--scan-yaml", "/opt/airflow/soda/checks.yml"])
+            print("✅ Soda scan completed successfully")
+        except subprocess.CalledProcessError as e:
+            if e.returncode == 1:  # which command failed - soda not found
+                print("⚠️ Soda not found, skipping data quality checks")
+                return
+            else:
+                print(f"Soda scan failed with return code {e.returncode}")
+                sys.exit(1)
+        except Exception as e:
+            print(f"Soda scan failed with error: {str(e)}")
+            sys.exit(1)
 
-    t4 = PythonOperator(task_id="data_quality", python_callable=soda_check)
+    # Add database schema initialization task
+    t0 = PostgresOperator(
+        task_id="init_schema",
+        postgres_conn_id="postgres_default",
+        sql="/opt/airflow/sql/schema_staging.sql"
+    )
+
+    t1 = PythonOperator(task_id="extract", python_callable=run_extract)
+    t2 = PythonOperator(task_id="load", python_callable=run_load)
+    t3 = PostgresOperator(
+        task_id="transform",
+        postgres_conn_id="postgres_default",
+        sql="/opt/airflow/sql/schema_core.sql"
+    )
+    t4 = PostgresOperator(
+        task_id="transform_data",
+        postgres_conn_id="postgres_default",
+        sql="/opt/airflow/scripts/transform.sql"
+    )
+    t5 = PythonOperator(task_id="data_quality", python_callable=soda_check)
 
-    t1 >> t2 >> t3 >> t4
+    t0 >> t1 >> t2 >> t3 >> t4 >> t5

infra/docker-compose.yml

@@ -1,62 +1,62 @@
-version: "3.8"
-services:
-  postgres:
-    image: postgres:14
-    environment:
-      POSTGRES_USER: airflow
-      POSTGRES_PASSWORD: airflow
-      POSTGRES_DB: youtube
-    ports:
-      - "5433:5432"
-    volumes:
-      - pgdata:/var/lib/postgresql/data
+# version: "3.8"
+# services:
+#   postgres:
+#     image: postgres:14
+#     environment:
+#       POSTGRES_USER: airflow
+#       POSTGRES_PASSWORD: airflow
+#       POSTGRES_DB: youtube
+#     ports:
+#       - "5433:5432"
+#     volumes:
+#       - pgdata:/var/lib/postgresql/data
 
-  webserver:
-    build:
-      context: ..
-      dockerfile: infra/Dockerfile.airflow
-    image: youtube-elt-airflow:local
-    environment:
-      AIRFLOW__CORE__EXECUTOR: LocalExecutor
-      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5433/youtube
-      AIRFLOW__CORE__LOAD_EXAMPLES: "False"
-      AIRFLOW__SECRETS__BACKEND: "airflow.secrets.environment_variables.EnvironmentVariablesBackend"
-    volumes:
-      - ./dags:/opt/airflow/dags
-      - ./scripts:/opt/airflow/scripts
-      - ./soda:/opt/airflow/soda
-      - ./sql:/opt/airflow/sql
-    ports:
-      - "8081:8080"
-    depends_on:
-      - postgres
-    command: webserver
+#   webserver:
+#     build:
+#       context: ..
+#       dockerfile: infra/Dockerfile.airflow
+#     image: youtube-elt-airflow:local
+#     environment:
+#       AIRFLOW__CORE__EXECUTOR: LocalExecutor
+#       AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5433/youtube
+#       AIRFLOW__CORE__LOAD_EXAMPLES: "False"
+#       AIRFLOW__SECRETS__BACKEND: "airflow.secrets.environment_variables.EnvironmentVariablesBackend"
+#     volumes:
+#       - ./dags:/opt/airflow/dags
+#       - ./scripts:/opt/airflow/scripts
+#       - ./soda:/opt/airflow/soda
+#       - ./sql:/opt/airflow/sql
+#     ports:
+#       - "58888:8080"
+#     depends_on:
+#       - postgres
+#     command: webserver
 
-  scheduler:
-    image: youtube-elt-airflow:local
-    volumes:
-      - ./dags:/opt/airflow/dags
-      - ./scripts:/opt/airflow/scripts
-      - ./soda:/opt/airflow/soda
-      - ./sql:/opt/airflow/sql
-    depends_on:
-      - webserver
-    command: scheduler
+#   scheduler:
+#     image: youtube-elt-airflow:local
+#     volumes:
+#       - ./dags:/opt/airflow/dags
+#       - ./scripts:/opt/airflow/scripts
+#       - ./soda:/opt/airflow/soda
+#       - ./sql:/opt/airflow/sql
+#     depends_on:
+#       - webserver
+#     command: scheduler
 
-  soda:
-    image: sodadata/soda-core:latest
-    entrypoint: ["tail", "-f", "/dev/null"]
-    depends_on:
-      - postgres
+#   soda:
+#     image: sodadata/soda-core:latest
+#     entrypoint: ["tail", "-f", "/dev/null"]
+#     depends_on:
+#       - postgres
 
-  streamlit:
-    image: python:3.10-slim
-    volumes:
-      - ./streamlit_app:/app
-    working_dir: /app
-    ports:
-      - "8501:8501"
-    command: ["bash", "-c", "pip install -r /app/../requirements.txt && streamlit run app.py --server.port 8501 --server.address 0.0.0.0"]
+#   streamlit:
+#     image: python:3.10-slim
+#     volumes:
+#       - ./streamlit_app:/app
+#     working_dir: /app
+#     ports:
+#       - "8501:8501"
+#     command: ["bash", "-c", "pip install -r /app/../requirements.txt && streamlit run app.py --server.port 8501 --server.address 0.0.0.0"]
 
-volumes:
-  pgdata:
+# volumes:
+#   pgdata:

pip.conf

@@ -0,0 +1,5 @@
+[global]
+timeout = 300
+retries = 5
+index-url = https://pypi.org/simple/
+extra-index-url = https://pip.astronomer.io/simple/

requirements-optional.txt

@@ -0,0 +1,7 @@
+# Optional dependencies that can be installed separately if needed
+soda-core>=3.5.5
+isodate>=0.6.1
+pytest>=7.4.4
+pytest-mock>=3.15.1
+flask>=2.2.5
+streamlit>=1.49.1

requirements.txt

@@ -1,8 +1,3 @@
-requests
-psycopg2-binary
-soda-core
-isodate
-pytest
-pytest-mock
-flask
-streamlit
+requests>=2.31.0
+psycopg2-binary>=2.9.9
+python-dotenv>=1.1.1

scripts/extract_youtube.py

@@ -1,7 +1,7 @@
 import os
 import requests
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -10,7 +10,7 @@
 CHANNEL_ID = os.getenv("YOUTUBE_CHANNEL_ID")
 
 # Crée un dossier "data" dans le projet et y enregistre le fichier
-DATA_DIR = "data"
+DATA_DIR = "/opt/airflow/data"
 os.makedirs(DATA_DIR, exist_ok=True)
 OUTPUT_PATH = os.path.join(DATA_DIR, "youtube_raw.json")
 
@@ -56,13 +56,24 @@ def get_videos_stats(video_ids, api_key):
     return results
 
 def main():
-    if not API_KEY or not CHANNEL_ID:
-        raise EnvironmentError("Set YOUTUBE_API_KEY and YOUTUBE_CHANNEL_ID in .env")
-    video_ids = list_videos_for_channel(CHANNEL_ID, API_KEY)
-    data = get_videos_stats(video_ids, API_KEY)
-    with open(OUTPUT_PATH, "w", encoding="utf-8") as f:
-        json.dump({"fetched_at": datetime.utcnow().isoformat(), "videos": data}, f, ensure_ascii=False, indent=2)
-    print(f"✅ Extracted {len(data)} videos to {OUTPUT_PATH}")
+    try:
+        if not API_KEY or not CHANNEL_ID:
+            raise EnvironmentError("Set YOUTUBE_API_KEY and YOUTUBE_CHANNEL_ID in .env")
+
+        print("Starting YouTube data extraction...")
+        video_ids = list_videos_for_channel(CHANNEL_ID, API_KEY)
+        print(f"Found {len(video_ids)} video IDs")
+
+        data = get_videos_stats(video_ids, API_KEY)
+        print(f"Retrieved stats for {len(data)} videos")
+
+        with open(OUTPUT_PATH, "w", encoding="utf-8") as f:
+            json.dump({"fetched_at": datetime.now(timezone.utc).isoformat(), "videos": data}, f, ensure_ascii=False, indent=2)
+        print(f"✅ Extracted {len(data)} videos to {OUTPUT_PATH}")
+
+    except Exception as e:
+        print(f"❌ Extraction failed: {str(e)}")
+        raise
 
 if __name__ == "__main__":
     main()

scripts/install_optional_deps.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+"""
+Script to install optional dependencies after the container is running.
+This helps avoid build-time timeouts.
+"""
+import subprocess
+import sys
+import os
+
+def install_optional_deps():
+    """Install optional dependencies from requirements-optional.txt"""
+    try:
+        requirements_file = "/opt/airflow/requirements-optional.txt"
+        if os.path.exists(requirements_file):
+            print("Installing optional dependencies...")
+            subprocess.check_call([
+                "pip", "install", 
+                "--timeout", "300",
+                "--retries", "5",
+                "-r", requirements_file
+            ])
+            print("✅ Optional dependencies installed successfully")
+        else:
+            print("⚠️ requirements-optional.txt not found, skipping optional dependencies")
+    except subprocess.CalledProcessError as e:
+        print(f"❌ Failed to install optional dependencies: {e}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"❌ Error installing optional dependencies: {e}")
+        sys.exit(1)
+
+if __name__ == "__main__":
+    install_optional_deps()