Data Orchestration: Airflow vs Dagster

Overview of Data Orchestration

What is a Pipeline Orchestrator?

Orchestrators manage, schedule, and monitor data workflows:

• Define workflows as directed acyclic graphs (DAGs)

• Schedule automated execution

• Handle dependencies and retries

• Monitor and alert on failures

• Manage resources and concurrency

Apache Airflow: The Industry Standard

Core Concepts

python

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='sales_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False,
) as dag:
    
    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract_function,
    )
    
    transform_task = PythonOperator(
        task_id='transform_data',
        python_callable=transform_function,
    )
    
    load_task = PythonOperator(
        task_id='load_data',
        python_callable=load_function,
    )
    
    extract_task >> transform_task >> load_task

Airflow Architecture

text

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Web Server    │────▶│    Scheduler    │────▶│    Executor     │
│  (UI + API)     │     │  (DAG parsing,  │     │  (Task exec)    │
│                 │     │   Scheduling)   │     │                 │
└─────────────────┘     └─────────────────┘     └─────────────────┘
         │                        │                        │
         │                        │                        │
         ▼                        ▼                        ▼
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Metadata DB   │◀────│    Workers      │◀────│   Message Queue │
│  (PostgreSQL/   │     │  (Celery/K8s)   │     │  (Redis/RabbitMQ)│
│   MySQL)        │     │                 │     │                 │
└─────────────────┘     └─────────────────┘     └─────────────────┘

Key Features

1. Dynamic DAG Generation

python

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator

def generate_dags():
    """Generate multiple similar DAGs dynamically"""
    for region in ['us', 'eu', 'apac']:
        
        with DAG(
            dag_id=f'sales_pipeline_{region}',
            schedule_interval='@daily',
            default_args=default_args,
        ) as dag:
            
            start = DummyOperator(task_id='start')
            
            process = PythonOperator(
                task_id=f'process_{region}',
                python_callable=lambda: process_region(region),
            )
            
            notify = PythonOperator(
                task_id=f'notify_{region}',
                python_callable=lambda: send_notification(region),
            )
            
            start >> process >> notify
        
        globals()[f'sales_pipeline_{region}'] = dag

2. TaskFlow API (Airflow 2.0+)

python

from airflow.decorators import dag, task
from datetime import datetime
import pandas as pd
import polars as pl

@dag(
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False,
)
def modern_data_pipeline():
    
    @task
    def extract() -> dict:
        """Extract data from multiple sources"""
        return {
            'sales': pl.read_csv('sales.csv'),
            'customers': pd.read_sql('customers', engine),
            'inventory': pl.read_parquet('inventory.parquet'),
        }
    
    @task
    def transform(sources: dict) -> pl.DataFrame:
        """Transform and join data"""
        sales = sources['sales']
        customers = pl.from_pandas(sources['customers'])
        
        transformed = (
            sales.lazy()
            .join(customers.lazy(), on='customer_id', how='left')
            .with_columns([
                (pl.col('price') * pl.col('quantity')).alias('revenue'),
                pl.col('date').str.strptime(pl.Date).alias('date_parsed'),
            ])
            .filter(pl.col('date_parsed').dt.year() == 2024)
            .collect()
        )
        
        return transformed
    
    @task
    def load(data: pl.DataFrame) -> str:
        """Load data to warehouse"""
        # Convert to pandas for SQLAlchemy compatibility
        data.to_pandas().to_sql(
            'sales_processed',
            engine,
            if_exists='replace',
            index=False
        )
        
        return f"Loaded {len(data)} rows"
    
    @task
    def validate(loaded_status: str):
        """Data quality validation"""
        from airflow.providers.postgres.hooks.postgres import PostgresHook
        
        hook = PostgresHook(postgres_conn_id='data_warehouse')
        df = hook.get_pandas_df("""
            SELECT 
                COUNT(*) as total_rows,
                COUNT(DISTINCT customer_id) as unique_customers,
                SUM(revenue) as total_revenue
            FROM sales_processed
            WHERE date >= CURRENT_DATE - INTERVAL '1 day'
        """)
        
        if df['total_revenue'].iloc[0] <= 0:
            raise ValueError("Revenue validation failed")
        
        return df.to_dict('records')
    
    # Define pipeline
    sources = extract()
    transformed = transform(sources)
    status = load(transformed)
    validation = validate(status)
    
    # Implicit dependencies via return values
    sources >> transformed >> status >> validation

pipeline = modern_data_pipeline()

3. XComs for Task Communication

python

from airflow.models import XCom

@task
def extract_large_data():
    """Store large data in external storage, reference in XCom"""
    import pickle
    from tempfile import NamedTemporaryFile
    
    large_df = pl.read_parquet('large_data.parquet')
    
    # Store in external file
    with NamedTemporaryFile(mode='wb', delete=False) as f:
        pickle.dump(large_df, f)
        filepath = f.name
    
    # Return reference in XCom
    return {'filepath': filepath, 'rows': len(large_df)}

@task
def process_large_data(data_ref: dict):
    """Retrieve large data from external storage"""
    import pickle
    
    with open(data_ref['filepath'], 'rb') as f:
        df = pickle.load(f)
    
    # Process data
    result = df.group_by('category').agg(pl.col('value').sum())
    
    # Cleanup
    import os
    os.remove(data_ref['filepath'])
    
    return result

Airflow for Data Engineering: Common Patterns

1. Multi-environment Pipeline

python

from airflow.models import Variable
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

class DataWarehousePipeline:
    def __init__(self, env='dev'):
        self.env = env
        self.s3_bucket = Variable.get(f's3_bucket_{env}')
        self.snowflake_schema = Variable.get(f'snowflake_schema_{env}')
    
    def create_dag(self):
        with DAG(
            dag_id=f'data_warehouse_ingestion_{self.env}',
            schedule_interval='@hourly',
        ) as dag:
            
            # Extract from multiple sources
            extract_s3 = PythonOperator(
                task_id='extract_from_s3',
                python_callable=self.extract_from_s3,
            )
            
            extract_api = PythonOperator(
                task_id='extract_from_api',
                python_callable=self.extract_from_api,
            )
            
            # Transform with Polars
            transform = PythonOperator(
                task_id='transform_data',
                python_callable=self.transform_with_polars,
            )
            
            # Load to Snowflake
            load = SnowflakeOperator(
                task_id='load_to_snowflake',
                sql='''
                    COPY INTO {{ params.table }}
                    FROM @{{ params.stage }}
                    FILE_FORMAT = (TYPE = 'PARQUET')
                ''',
                snowflake_conn_id=f'snowflake_{self.env}',
                params={
                    'table': f'{self.snowflake_schema}.sales',
                    'stage': f'{self.snowflake_schema}.s3_stage',
                }
            )
            
            # Data quality checks
            quality_check = PythonOperator(
                task_id='data_quality_checks',
                python_callable=self.run_quality_checks,
                trigger_rule='all_done',
            )
            
            # Set dependencies
            [extract_s3, extract_api] >> transform >> load >> quality_check
            
            return dag
    
    def extract_from_s3(self):
        s3_hook = S3Hook(aws_conn_id=f'aws_{self.env}')
        
        # List files in S3
        files = s3_hook.list_keys(
            bucket_name=self.s3_bucket,
            prefix='sales/raw/'
        )
        
        # Download and process with Polars
        for file in files[:10]:  # Process first 10 files
            local_path = s3_hook.download_file(
                key=file,
                bucket_name=self.s3_bucket,
            )
            
            df = pl.read_parquet(local_path)
            # Process dataframe...
            
        return len(files)
    
    def transform_with_polars(self, **context):
        ti = context['ti']
        s3_count = ti.xcom_pull(task_ids='extract_from_s3')
        api_count = ti.xcom_pull(task_ids='extract_from_api')
        
        # Use Polars for efficient transformation
        all_data = pl.DataFrame()
        
        for i in range(s3_count):
            # Process S3 data with lazy evaluation
            df = (
                pl.scan_parquet(f'/tmp/s3_data_{i}.parquet')
                .filter(pl.col('value') > 0)
                .select([
                    pl.col('id'),
                    pl.col('timestamp').cast(pl.Date),
                    (pl.col('price') * pl.col('quantity')).alias('revenue'),
                ])
                .collect()
            )
            all_data = pl.concat([all_data, df])
        
        # Perform aggregations
        aggregated = (
            all_data.lazy()
            .group_by(pl.col('timestamp').dt.strftime('%Y-%m'))
            .agg([
                pl.col('revenue').sum().alias('monthly_revenue'),
                pl.col('id').n_unique().alias('unique_customers'),
            ])
            .sort('timestamp')
            .collect()
        )
        
        # Save to intermediate location
        aggregated.write_parquet(f'/tmp/transformed_{self.env}.parquet')
        return aggregated.shape

Airflow Pros & Cons

Pros:

• Mature and battle-tested

• Large community and ecosystem

• Rich UI with task monitoring

• Excellent scheduling capabilities

• Extensive provider ecosystem

Cons:

• Steep learning curve

• Heavyweight for simple pipelines

• DAG parsing can be slow

• Limited data-aware scheduling

• Complex local development setup

Dagster: The Modern Data Orchestrator

Core Philosophy

• Software-defined assets

• Development productivity focus

• Type system and testing

• Data dependency tracking

Basic Concepts

python

from dagster import Definitions, asset, job, op, graph, AssetKey
import pandas as pd
import polars as pl
from typing import Dict, List

@asset
def raw_sales_data() -> pl.DataFrame:
    """Load raw sales data"""
    return pl.read_csv('sales.csv')

@asset
def raw_customer_data() -> pl.DataFrame:
    """Load raw customer data"""
    return pl.read_csv('customers.csv')

@asset
def cleaned_sales_data(raw_sales_data: pl.DataFrame) -> pl.DataFrame:
    """Clean and validate sales data"""
    return (
        raw_sales_data
        .filter(pl.col('amount') > 0)
        .with_columns(
            pl.col('timestamp').str.strptime(pl.Datetime).alias('datetime')
        )
        .drop_nulls()
    )

@asset
def sales_analysis(
    cleaned_sales_data: pl.DataFrame,
    raw_customer_data: pl.DataFrame
) -> Dict:
    """Analyze sales with customer data"""
    
    # Join with customer data
    analysis = (
        cleaned_sales_data.lazy()
        .join(
            raw_customer_data.lazy(),
            left_on='customer_id',
            right_on='id',
            how='left'
        )
        .group_by('region', 'product_category')
        .agg([
            pl.col('amount').sum().alias('total_sales'),
            pl.col('customer_id').n_unique().alias('unique_customers'),
            pl.col('amount').mean().alias('avg_transaction'),
        ])
        .collect()
    )
    
    # Generate insights
    top_regions = (
        analysis.sort('total_sales', descending=True)
        .head(5)
        .to_dicts()
    )
    
    return {
        'analysis_df': analysis,
        'top_regions': top_regions,
        'summary': {
            'total_rows': len(analysis),
            'total_sales': analysis['total_sales'].sum(),
            'avg_customers': analysis['unique_customers'].mean(),
        }
    }

@asset
def sales_report(sales_analysis: Dict) -> str:
    """Generate sales report from analysis"""
    import json
    from datetime import datetime
    
    report = {
        'generated_at': datetime.now().isoformat(),
        'summary': sales_analysis['summary'],
        'top_performing_regions': sales_analysis['top_regions'],
    }
    
    # Save report
    report_path = f'reports/sales_{datetime.now().strftime("%Y%m%d")}.json'
    with open(report_path, 'w') as f:
        json.dump(report, f, indent=2)
    
    return report_path

# Define the data pipeline
defs = Definitions(
    assets=[
        raw_sales_data,
        raw_customer_data,
        cleaned_sales_data,
        sales_analysis,
        sales_report,
    ]
)

Key Features

1. Software-Defined Assets

python

from dagster import (
    asset,
    AssetIn,
    AssetOut,
    multi_asset,
    Output,
    AssetExecutionContext
)
from dagster_polars import PolarsDataFrame

@multi_asset(
    ins={
        "raw_data": AssetIn(key="raw_sales_data")
    },
    outs={
        "cleaned_data": AssetOut(io_manager_key="parquet_io"),
        "validation_report": AssetOut(io_manager_key="json_io"),
    }
)
def clean_and_validate(
    context: AssetExecutionContext,
    raw_data: pl.DataFrame
) -> tuple[pl.DataFrame, dict]:
    """Clean data and produce validation report"""
    
    # Data cleaning
    cleaned = (
        raw_data
        .filter(pl.col('amount').is_not_null())
        .with_columns(
            pl.when(pl.col('amount') < 0)
            .then(0)
            .otherwise(pl.col('amount'))
            .alias('amount_positive')
        )
        .drop('amount')
        .rename({'amount_positive': 'amount'})
    )
    
    # Data quality checks
    validation = {
        'rows_processed': len(cleaned),
        'null_values': cleaned.null_count().to_dict(),
        'value_ranges': {
            'amount': {
                'min': float(cleaned['amount'].min()),
                'max': float(cleaned['amount'].max()),
                'mean': float(cleaned['amount'].mean()),
            }
        },
        'issues_found': len(raw_data) - len(cleaned),
    }
    
    context.log.info(f"Validation: {validation}")
    
    yield Output(cleaned, output_name="cleaned_data")
    yield Output(validation, output_name="validation_report")

2. Type System and Testing

python

from dagster import Config, OpExecutionContext
from pydantic import Field
import pandas as pd

class ProcessConfig(Config):
    """Configuration schema for processing"""
    region: str = Field(description="Region to process")
    start_date: str = Field(description="Start date in YYYY-MM-DD")
    end_date: str = Field(description="End date in YYYY-MM-DD")
    threshold: float = Field(default=1000.0, description="Revenue threshold")

@asset
def regional_sales(context: OpExecutionContext, config: ProcessConfig) -> pl.DataFrame:
    """Process sales for a specific region with configuration"""
    context.log.info(f"Processing region: {config.region}")
    
    # Load and filter data
    df = pl.read_parquet('sales.parquet')
    
    filtered = (
        df.lazy()
        .filter(
            (pl.col('region') == config.region) &
            (pl.col('date') >= config.start_date) &
            (pl.col('date') <= config.end_date) &
            (pl.col('revenue') >= config.threshold)
        )
        .collect()
    )
    
    return filtered

# Test the asset
def test_regional_sales():
    """Unit test for the regional_sales asset"""
    from dagster import build_asset_context
    
    context = build_asset_context()
    config = ProcessConfig(
        region='US',
        start_date='2024-01-01',
        end_date='2024-01-31',
        threshold=500.0
    )
    
    result = regional_sales(context, config)
    
    # Assertions
    assert len(result) > 0
    assert (result['region'] == 'US').all()
    assert (result['revenue'] >= 500.0).all()

3. Data Dependencies & Partitioning

python

from dagster import (
    asset,
    DailyPartitionsDefinition,
    StaticPartitionsDefinition,
    multi_asset,
    AssetIn
)

# Time-based partitioning
daily_partitions = DailyPartitionsDefinition(start_date="2024-01-01")

@asset(partitions_def=daily_partitions)
def daily_sales_aggregate(context) -> pl.DataFrame:
    """Process sales data for a specific day"""
    partition_date = context.asset_partition_key_for_output()
    
    # Load data for the specific day
    df = pl.read_parquet(f'sales/daily/{partition_date}.parquet')
    
    aggregated = (
        df.lazy()
        .group_by('product_id', 'region')
        .agg([
            pl.col('amount').sum().alias('daily_sales'),
            pl.col('customer_id').n_unique().alias('daily_customers'),
            pl.col('amount').mean().alias('avg_transaction'),
        ])
        .collect()
    )
    
    return aggregated

# Multi-dimensional partitioning
region_partitions = StaticPartitionsDefinition(['us', 'eu', 'apac'])
product_partitions = StaticPartitionsDefinition(['electronics', 'clothing', 'food'])

@asset(partitions_def=[region_partitions, product_partitions])
def regional_product_sales(context) -> Dict:
    """Process sales by region and product category"""
    region = context.partition_key.keys_by_dimension['region']
    product = context.partition_key.keys_by_dimension['product']
    
    # Load and filter
    df = pl.read_parquet('sales.parquet')
    
    filtered = df.filter(
        (pl.col('region') == region) &
        (pl.col('product_category') == product)
    )
    
    analysis = filtered.group_by('date').agg(
        pl.col('amount').sum().alias('daily_revenue'),
        pl.col('customer_id').n_unique().alias('daily_customers'),
    ).sort('date')
    
    return {
        'region': region,
        'product': product,
        'data': analysis,
        'summary': {
            'total_revenue': analysis['daily_revenue'].sum(),
            'avg_daily_customers': analysis['daily_customers'].mean(),
            'days_analyzed': len(analysis),
        }
    }

Dagster with Polars: Advanced Patterns

python

from dagster import (
    Definitions,
    asset,
    MaterializeResult,
    MetadataValue,
    AssetObservation
)
from dagster_polars import PolarsDataFrame
import polars as pl

class PolarsDataPipeline:
    """Advanced data pipeline with Polars and Dagster"""
    
    @asset(
        compute_kind="polars",
        io_manager_key="parquet_io_manager"
    )
    def process_large_dataset(self, context) -> PolarsDataFrame:
        """Process large dataset with lazy evaluation"""
        context.log.info("Starting large dataset processing...")
        
        # Use streaming for out-of-core processing
        result = (
            pl.scan_parquet('/data/large/*.parquet')
            .filter(pl.col('year') == 2024)
            .select([
                'id',
                'date',
                'region',
                'revenue',
                'cost',
                (pl.col('revenue') - pl.col('cost')).alias('profit')
            ])
            .group_by(['region', 'date'])
            .agg([
                pl.col('revenue').sum().alias('total_revenue'),
                pl.col('cost').sum().alias('total_cost'),
                pl.col('profit').sum().alias('total_profit'),
                pl.col('id').n_unique().alias('unique_customers'),
            ])
            .sort(['region', 'date'])
            .collect(streaming=True)  # Stream processing
        )
        
        # Add metadata
        return MaterializeResult(
            metadata={
                "rows_processed": MetadataValue.int(len(result)),
                "columns": MetadataValue.json(list(result.columns)),
                "memory_usage_mb": MetadataValue.float(result.estimated_size() / 1024 / 1024),
                "sample_data": MetadataValue.md(result.head().to_pandas().to_markdown()),
            }
        )
    
    @asset
    def validate_data_quality(
        self,
        context,
        process_large_dataset: PolarsDataFrame
    ) -> MaterializeResult:
        """Validate data quality metrics"""
        df = process_large_dataset
        
        # Calculate quality metrics
        quality_metrics = {
            'total_rows': len(df),
            'null_counts': df.null_count().to_dict(),
            'negative_profits': df.filter(pl.col('total_profit') < 0).height,
            'zero_revenue': df.filter(pl.col('total_revenue') == 0).height,
            'duplicate_keys': df.group_by(['region', 'date']).agg(
                pl.count().alias('count')
            ).filter(pl.col('count') > 1).height,
        }
        
        # Check quality thresholds
        quality_issues = []
        if quality_metrics['negative_profits'] > 0:
            quality_issues.append("Negative profits detected")
        if quality_metrics['duplicate_keys'] > 0:
            quality_issues.append("Duplicate region/date combinations")
        
        # Record observations
        context.add_output_metadata({
            "quality_metrics": MetadataValue.json(quality_metrics),
            "quality_issues": MetadataValue.json(quality_issues),
            "quality_score": MetadataValue.float(
                100 - len(quality_issues) * 10  # Simple scoring
            ),
        })
        
        # Yield observation for monitoring
        yield AssetObservation(
            asset_key="validate_data_quality",
            metadata={
                "validation_passed": MetadataValue.bool(len(quality_issues) == 0),
                "issues_count": MetadataValue.int(len(quality_issues)),
            }
        )
        
        if quality_issues:
            raise ValueError(f"Quality check failed: {quality_issues}")
        
        return MaterializeResult(
            metadata={
                "validation_status": "PASSED",
                "issues_found": MetadataValue.int(len(quality_issues)),
            }
        )
    
    @asset(deps=[validate_data_quality])
    def generate_analytics_report(
        self,
        context,
        process_large_dataset: PolarsDataFrame
    ) -> str:
        """Generate analytical report from processed data"""
        df = process_large_dataset
        
        # Perform advanced analytics
        analytics = {
            'top_regions': (
                df.group_by('region')
                .agg(pl.col('total_profit').sum().alias('region_profit'))
                .sort('region_profit', descending=True)
                .head(5)
                .to_dicts()
            ),
            'profit_trend': (
                df.group_by('date')
                .agg(pl.col('total_profit').sum().alias('daily_profit'))
                .sort('date')
                .to_dicts()
            ),
            'profitability_ratio': (
                df.with_columns(
                    (pl.col('total_profit') / pl.col('total_revenue')).alias('profit_margin')
                )
                .select('profit_margin')
                .mean()
                .to_dicts()[0]['profit_margin']
            ),
        }
        
        # Generate report
        report = f"""
        # Sales Analytics Report
        ## Summary
        - Total Rows Analyzed: {len(df):,}
        - Average Profit Margin: {analytics['profitability_ratio']:.2%}
        - Time Period: {df['date'].min()} to {df['date'].max()}
        
        ## Top Performing Regions
        {pl.DataFrame(analytics['top_regions']).to_pandas().to_markdown()}
        """
        
        # Save report
        report_path = f"/reports/analytics_{context.run_id}.md"
        with open(report_path, 'w') as f:
            f.write(report)
        
        context.add_output_metadata({
            "report_path": MetadataValue.path(report_path),
            "analytics_summary": MetadataValue.json({
                'top_regions_count': len(analytics['top_regions']),
                'trend_points': len(analytics['profit_trend']),
                'avg_profit_margin': analytics['profitability_ratio'],
            })
        })
        
        return report_path

# Define the pipeline
defs = Definitions(
    assets=[
        PolarsDataPipeline().process_large_dataset,
        PolarsDataPipeline().validate_data_quality,
        PolarsDataPipeline().generate_analytics_report,
    ],
    resources={
        "parquet_io_manager": ParquetIOManager(),  # Custom IO manager
    }
)

Comparison: Airflow vs Dagster

Architectural Differences

Performance Comparison

python

# AIRFLOW: Traditional DAG-based scheduling
# Pros: Fine-grained control, complex schedules
# Cons: Can't skip unnecessary tasks based on data

# DAGSTER: Asset-based with incremental computation
# Pros: Skip processing if data unchanged
# Cons: Less control over exact execution timing

Use Case Recommendations

Choose Airflow when:

• You need complex scheduling (cron expressions, etc.)

• You have existing Airflow infrastructure

• Your team is already proficient with Airflow

• You need extensive third-party integrations

• Operations/reliability is the primary concern

Choose Dagster when:

• Data dependencies are more important than time schedules

• You want better developer experience

• You need strong typing and testing

• Your pipeline logic is complex with many data assets

• You're building a new data platform from scratch

Hybrid Approach: Using Both

Airflow for Orchestration, Dagster for Data Logic

python

# dagster_data_logic.py
from dagster import Definitions, asset, OpExecutionContext
import polars as pl

@asset
def process_data(context: OpExecutionContext) -> pl.DataFrame:
    """Complex data processing logic"""
    context.log.info("Processing data with Polars...")
    
    df = pl.read_parquet('/data/raw.parquet')
    
    processed = (
        df.lazy()
        .filter(pl.col('value') > 0)
        .with_columns([
            (pl.col('price') * pl.col('quantity')).alias('revenue'),
            pl.col('date').str.strptime(pl.Date).alias('date_parsed'),
        ])
        .group_by(['date_parsed', 'category'])
        .agg([
            pl.col('revenue').sum().alias('daily_revenue'),
            pl.col('customer_id').n_unique().alias('unique_customers'),
        ])
        .collect()
    )
    
    return processed

# airflow_orchestration.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.dagster.operators.dagster import DagsterOperator

def create_hybrid_dag():
    with DAG('hybrid_pipeline', schedule_interval='@daily') as dag:
        
        # Use Dagster for complex data processing
        data_processing = DagsterOperator(
            task_id='process_data_with_dagster',
            dagster_conn_id='dagster_default',
            repository_name='data_repo',
            op_name='process_data',
        )
        
        # Use Airflow for operational tasks
        operational_task = PythonOperator(
            task_id='send_alerts',
            python_callable=send_alert_function,
        )
        
        data_processing >> operational_task
    
    return dag

For Airflow:

1. Use TaskFlow API for Python-heavy pipelines

2. Implement retries with exponential backoff

3. Use Variables and Connections for configuration

4. Monitor resource usage with StatsD/Prometheus

5. Implement proper error handling and alerts

6. Use Pools and Priority Weights for resource management

For Dagster:

1. Leverage the type system for validation

2. Use partitions for incremental processing

3. Implement comprehensive tests for assets

4. Use metadata for observability

5. Configure appropriate IO managers

6. Use asset dependencies instead of explicit task ordering

General Data Engineering Practices:

1. Idempotency: Pipelines should be rerunnable

2. Observability: Log everything, add metrics

3. Data Quality: Build validation into pipelines

4. Incremental Processing: Process only new/changed data

5. Resource Management: Handle memory and compute constraints

6. Testing: Test data transformations thoroughly

Modern Stack Example

python

# Complete modern data stack
"""
┌─────────────────────────────────────────────────────────────┐
│                   Modern Data Stack                          │
├─────────────────────────────────────────────────────────────┤
│  Dagster (Orchestration) ──────┬───── Airflow (Scheduling)  │
├─────────────────────────────────┼────────────────────────────┤
│  Polars (Transformation) ───────┼───── Pandas (Analysis)    │
├─────────────────────────────────┼────────────────────────────┤
│  DuckDB (Analytical Queries)    │  Great Expectations (DQ)  │
├─────────────────────────────────┼────────────────────────────┤
│  Parquet/Delta (Storage) ───────┼───── PostgreSQL (Metadata)│
└─────────────────────────────────┴────────────────────────────┘
"""

class ModernDataPlatform:
    """Example of a modern data platform combining tools"""
    
    def __init__(self):
        # Use Dagster for asset management
        self.orchestrator = DagsterOrchestrator()
        
        # Use Airflow for complex scheduling if needed
        self.scheduler = AirflowScheduler()
        
        # Use Polars for ETL
        self.transformer = PolarsTransformer()
        
        # Use DuckDB for analytical queries
        self.analyzer = DuckDBAnalyzer()
    
    def build_pipeline(self):
        """Build a complete data pipeline"""
        @asset
        def extract_from_sources():
            """Extract from multiple sources"""
            # Use Polars for efficient data loading
            s3_data = pl.read_parquet('s3://bucket/data.parquet')
            api_data = pl.from_pandas(pd.read_json('api_endpoint'))
            db_data = pl.read_database('SELECT * FROM table', connection)
            
            return {'s3': s3_data, 'api': api_data, 'db': db_data}
        
        @asset
        def transform_with_polars(sources: dict) -> pl.DataFrame:
            """Transform using Polars lazy evaluation"""
            combined = pl.concat([
                sources['s3'],
                sources['api'],
                sources['db'],
            ])
            
            transformed = (
                combined.lazy()
                .filter(pl.col('is_valid') == True)
                .with_columns([
                    (pl.col('revenue') - pl.col('cost')).alias('profit'),
                    pl.col('timestamp').dt.truncate('1d').alias('date'),
                ])
                .group_by(['date', 'region', 'product'])
                .agg([
                    pl.col('profit').sum().alias('daily_profit'),
                    pl.col('customer_id').n_unique().alias('customers'),
                ])
                .collect()
            )
            
            return transformed
        
        @asset
        def analyze_with_duckdb(transformed: pl.DataFrame):
            """Use DuckDB for complex analytical queries"""
            import duckdb
            
            # Register Polars DataFrame with DuckDB
            duckdb.register('sales', transformed.to_arrow())
            
            # Run analytical queries
            analysis = duckdb.sql("""
                WITH daily_stats AS (
                    SELECT 
                        date,
                        region,
                        SUM(daily_profit) as total_profit,
                        SUM(customers) as total_customers,
                        AVG(daily_profit) as avg_daily_profit
                    FROM sales
                    GROUP BY date, region
                ),
                moving_avg AS (
                    SELECT
                        date,
                        region,
                        total_profit,
                        AVG(total_profit) OVER (
                            PARTITION BY region 
                            ORDER BY date 
                            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
                        ) as weekly_avg_profit
                    FROM daily_stats
                )
                SELECT *
                FROM moving_avg
                ORDER BY date DESC, total_profit DESC
            """).pl()  # Convert back to Polars
            
            return analysis
        
        @asset
        def validate_with_gx(analysis: pl.DataFrame):
            """Data quality validation with Great Expectations"""
            import great_expectations as gx
            
            context = gx.get_context()
            
            # Create expectation suite
            suite = context.add_expectation_suite("sales_validation")
            
            # Define expectations
            expectations = [
                {"expectation_type": "expect_column_values_to_not_be_null", "column": "total_profit"},
                {"expectation_type": "expect_column_values_to_be_between", 
                 "column": "total_profit", "min_value": 0},
                {"expectation_type": "expect_column_pair_values_a_to_be_greater_than_b",
                 "column_A": "total_profit", "column_B": "weekly_avg_profit", "or_equal": True},
            ]
            
            # Add expectations
            for exp in expectations:
                suite.add_expectation(exp)
            
            # Validate data
            validator = context.get_validator(
                batch_request={
                    "datasource_name": "polars_datasource",
                    "data_connector_name": "default_inferred_data_connector_name",
                    "data_asset_name": "sales_analysis",
                    "data": analysis.to_pandas(),  # GE works with pandas
                },
                expectation_suite=suite,
            )
            
            results = validator.validate()
            return results.to_json_dict()
        
        # Return the complete pipeline
        return Definitions(
            assets=[
                extract_from_sources,
                transform_with_polars,
                analyze_with_duckdb,
                validate_with_gx,
            ]
        )