Data Pipelines

📊 ETL/ELT Pipeline Stages

📥 Betting Data Sources

🖥️ Pipeline Status

✅ Best Practices

Python / Airflow Example

# Airflow DAG for odds ingestion
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = { 
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

def ingest_odds():
    """Fetch odds from API and store in raw layer"""
    import requests
    response = requests.get("https://api.oddsapi.com/v4/sports/...")
    # Store to S3/GCS
    
def transform_odds():
    """Clean and standardize odds data"""
    # dbt run --select odds_staging
    
def validate_data():
    """Run data quality checks"""
    # Check for nulls, outliers, staleness

with DAG('odds_pipeline', 
         schedule_interval='*/5 * * * *',  # Every 5 min
         default_args=default_args) as dag:
    
    ingest = PythonOperator(task_id='ingest', python_callable=ingest_odds)
    transform = PythonOperator(task_id='transform', python_callable=transform_odds)
    validate = PythonOperator(task_id='validate', python_callable=validate_data)
    
    ingest >> transform >> validate