StreamCart Airflow Project

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator import json import random default_args = { 'owner': 'streamcart', 'depends_on_past': False, 'email_on_failure': False, 'email_on_retry': False, 'retries': 2, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'ingest_web_events', default_args=default_args, description='Ingest web clickstream data every 15 minutes', schedule='*/15 * * * *', # Every 15 minutes start_date=datetime(2024, 1, 1), catchup=False, tags=['streamcart', 'ingestion', 'realtime'], )

def fetch_events(**context): """Simulate fetching events from web analytics API""" events = [] for i in range(random.randint(10, 50)): event = { 'event_id': f'evt_{i}_{datetime.now().timestamp()}', 'user_id': f'user_{random.randint(1, 1000)}', 'event_type': random.choice(['page_view', 'add_to_cart', 'purchase']), 'product_id': f'prod_{random.randint(1, 100)}', 'timestamp': datetime.now().isoformat() } events.append(event) # Save to temp file with open('/tmp/web_events.json', 'w') as f: json.dump(events, f) print(f"Fetched {len(events)} events") return len(events) def validate_events(**context): """Validate fetched events""" with open('/tmp/web_events.json', 'r') as f: events = json.load(f) assert len(events) > 0, "No events fetched" assert all('event_id' in e for e in events), "Missing event_id" print(f"Validated {len(events)} events") return True def load_to_warehouse(**context): """Load events to data warehouse (simulated)""" with open('/tmp/web_events.json', 'r') as f: events = json.load(f) # Simulate loading to warehouse print(f"Loading {len(events)} events to warehouse...") # In real scenario: INSERT INTO events_table VALUES ... print(f"✓ Loaded {len(events)} events successfully") return len(events)

# Create tasks fetch_task = PythonOperator( task_id='fetch_web_events', python_callable=fetch_events, dag=dag, ) validate_task = PythonOperator( task_id='validate_events', python_callable=validate_events, dag=dag, ) load_task = PythonOperator( task_id='load_to_warehouse', python_callable=load_to_warehouse, dag=dag, ) cleanup_task = BashOperator( task_id='cleanup_temp_files', bash_command='rm -f /tmp/web_events.json', dag=dag, ) # Set dependencies fetch_task >> validate_task >> load_task >> cleanup_task

from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator import random default_args = { 'owner': 'streamcart', 'depends_on_past': False, 'email_on_failure': True, 'email': ['data-team@streamcart.com'], 'retries': 1, 'retry_delay': timedelta(minutes=10), } dag = DAG( 'daily_product_summary', default_args=default_args, description='Daily aggregation of product metrics', schedule='0 2 * * *', # Daily at 2 AM start_date=datetime(2024, 1, 1), catchup=False, tags=['streamcart', 'analytics', 'daily'], ) def extract_daily_events(**context): """Extract yesterday's events""" execution_date = context['execution_date'] print(f"Extracting events for {execution_date.date()}") # Simulate extraction return {'events_count': random.randint(1000, 5000)} def calculate_metrics(**context): """Calculate product metrics""" ti = context['ti'] data = ti.xcom_pull(task_ids='extract_events') metrics = { 'total_views': data['events_count'], 'total_purchases': int(data['events_count'] * 0.05), 'conversion_rate': 0.05, 'top_products': [f'prod_{i}' for i in range(1, 11)] } print(f"Calculated metrics: {metrics}") return metrics def generate_report(**context): """Generate business report""" ti = context['ti'] metrics = ti.xcom_pull(task_ids='calculate_metrics') report = f""" Daily Product Summary ==================== Date: {context['execution_date'].date()} Total Views: {metrics['total_views']:,} Total Purchases: {metrics['total_purchases']:,} Conversion Rate: {metrics['conversion_rate']:.1%} Top Products: {', '.join(metrics['top_products'][:5])} """ print(report) return report # Create tasks extract = PythonOperator( task_id='extract_events', python_callable=extract_daily_events, dag=dag, ) calculate = PythonOperator( task_id='calculate_metrics', python_callable=calculate_metrics, dag=dag, ) report = PythonOperator( task_id='generate_report', python_callable=generate_report, dag=dag, ) extract >> calculate >> report