267 Automation 4db36136

Automation Patterns Reference

Pattern Summary

Modern Recommendation: Use AutomationCondition for new asset-centric pipelines. Schedules and sensors remain useful for ops-based jobs and event-driven automation.

Declarative Automation

Modern automation pattern that sets conditions on assets rather than scheduling jobs.

What is Declarative Automation?

Instead of defining schedules or sensors that trigger jobs, you define conditions directly on assets. Dagster evaluates these conditions and automatically materializes assets when conditions are met.

Basic AutomationCondition Patterns

import dagster as dg
from dagster import AutomationCondition

# Materialize when missing (never been created)
@dg.asset(
    automation_condition=AutomationCondition.on_missing()
)
def initial_load() -> None:
    pass

# Materialize on cron schedule
@dg.asset(
    automation_condition=AutomationCondition.on_cron("0 9 * * *")
)
def daily_report() -> None:
    pass

# Materialize when upstream assets update
@dg.asset(
    deps=["raw_data"],
    automation_condition=AutomationCondition.eager()
)
def processed_data() -> None:
    """Automatically updates when raw_data changes."""
    pass

Combining Conditions

# OR: Materialize if ANY condition is true
@dg.asset(
    automation_condition=(
        AutomationCondition.on_missing()
        | AutomationCondition.on_cron("0 0 * * *")
    )
)
def flexible_asset() -> None:
    """Updates if missing OR daily at midnight."""
    pass

# AND: Materialize only if ALL conditions are true
@dg.asset(
    automation_condition=(
        AutomationCondition.on_cron("0 9 * * *")
        & AutomationCondition.any_deps_updated()
    )
)
def conditional_report() -> None:
    """Only updates at 9am if upstream deps changed."""
    pass

Common AutomationCondition Patterns

Benefits of Declarative Automation

1. Asset-Native: No separate job definitions needed
2. Expressive: Combine conditions with boolean logic
3. Dependency-Aware: Automatically understands asset graph
4. Better for Complex Logic: More readable than schedule/sensor combinations
5. Centralized: Automation logic lives with asset definition

When to Use

• Declarative Automation: Asset-centric pipelines with condition-based triggers
• Schedules: Simple time-based triggers, ops-based jobs, or when you need exact cron control
• Sensors: Event-driven automation with external systems, complex state tracking

Jobs

Jobs select which assets to materialize together:

Basic Job Definition

import dagster as dg

trip_update_job = dg.define_asset_job(
    name="trip_update_job",
    selection=["taxi_trips", "taxi_zones"],
)

Asset Selection Patterns

# Select specific assets
dg.AssetSelection.assets("asset_a", "asset_b")

# Select all assets
dg.AssetSelection.all()

# Select by group
dg.AssetSelection.groups("analytics")

# Select with dependencies
dg.AssetSelection.assets("final_report").upstream()  # Include upstream
dg.AssetSelection.assets("raw_data").downstream()    # Include downstream

# Combine selections
dg.AssetSelection.all() - dg.AssetSelection.assets("excluded_asset")
dg.AssetSelection.groups("a") | dg.AssetSelection.groups("b")  # Union
dg.AssetSelection.groups("a") & dg.AssetSelection.assets("specific")  # Intersection

Job with Tags

daily_job = dg.define_asset_job(
    name="daily_job",
    selection=dg.AssetSelection.all(),
    tags={"team": "data-eng", "priority": "high"},
)

Schedules

Basic Schedule

import dagster as dg
from my_project.defs.jobs import trip_update_job

trip_update_schedule = dg.ScheduleDefinition(
    job=trip_update_job,
    cron_schedule="0 0 5 * *",  # 5th of each month at midnight
)

Common Cron Patterns

Tip: Use Crontab Guru to create and test cron expressions.

Schedule with Timezone

my_schedule = dg.ScheduleDefinition(
    job=my_job,
    cron_schedule="0 9 * * *",  # 9:00 AM
    execution_timezone="America/New_York",
)

Custom Schedule with Decorator

@dg.schedule(cron_schedule="0 0 * * *", job=my_job)
def custom_schedule(context: dg.ScheduleEvaluationContext):
    """Schedule with custom logic."""
    scheduled_date = context.scheduled_execution_time.strftime("%Y-%m-%d")

    return dg.RunRequest(
        run_key=f"daily_{scheduled_date}",
        run_config={
            "ops": {
                "my_asset": {
                    "config": {"date": scheduled_date}
                }
            }
        },
    )

Partitioned Schedule

Automatically run for new partitions:

from my_project.defs.partitions import monthly_partition
from my_project.defs.jobs import partitioned_job

monthly_schedule = dg.build_schedule_from_partitioned_job(
    job=partitioned_job,
    description="Materializes data for the previous month",
)

Sensors

Sensor Anatomy

Sensors follow this lifecycle:

1. Read cursor (previous state)
2. Observe current state
3. Compare states and create run requests for changes
4. Update cursor

Basic Sensor Pattern

import dagster as dg
import json

@dg.sensor(job=my_job)
def file_sensor(context: dg.SensorEvaluationContext):
    # 1. Read cursor (previous state)
    previous_state = json.loads(context.cursor) if context.cursor else {}
    current_state = {}
    runs_to_request = []

    # 2. Observe current state
    for filepath in get_files_to_watch():
        last_modified = os.path.getmtime(filepath)
        filename = os.path.basename(filepath)
        current_state[filename] = last_modified

        # 3. Check for changes
        if filename not in previous_state or previous_state[filename] != last_modified:
            runs_to_request.append(dg.RunRequest(
                run_key=f"file_{filename}_{last_modified}",
                run_config={
                    "ops": {
                        "process_file": {
                            "config": {"filename": filename}
                        }
                    }
                }
            ))

    # 4. Return result with updated cursor
    return dg.SensorResult(
        run_requests=runs_to_request,
        cursor=json.dumps(current_state),
    )

File Sensor

import os
import json

@dg.sensor(job=adhoc_request_job)
def adhoc_request_sensor(context: dg.SensorEvaluationContext):
    PATH_TO_REQUESTS = os.path.join(
        os.path.dirname(__file__),
        "../../../data/requests",
    )

    previous_state = json.loads(context.cursor) if context.cursor else {}
    current_state = {}
    runs_to_request = []

    for filename in os.listdir(PATH_TO_REQUESTS):
        file_path = os.path.join(PATH_TO_REQUESTS, filename)
        if filename.endswith(".json") and os.path.isfile(file_path):
            last_modified = os.path.getmtime(file_path)
            current_state[filename] = last_modified

            if filename not in previous_state or previous_state[filename] != last_modified:
                with open(file_path, "r") as f:
                    request_config = json.load(f)

                runs_to_request.append(dg.RunRequest(
                    run_key=f"request_{filename}_{last_modified}",
                    run_config={
                        "ops": {
                            "adhoc_request": {
                                "config": {
                                    "filename": filename,
                                    **request_config
                                }
                            }
                        }
                    }
                ))

    return dg.SensorResult(
        run_requests=runs_to_request,
        cursor=json.dumps(current_state),
    )

Asset Sensor

Trigger when another asset materializes:

@dg.asset_sensor(asset_key=dg.AssetKey("upstream_asset"), job=downstream_job)
def upstream_sensor(context: dg.SensorEvaluationContext, asset_event):
    """Triggers when upstream_asset is materialized."""
    return dg.RunRequest(
        run_key=f"downstream_{asset_event.dagster_event.event_specific_data.materialization.run_id}",
    )

Sensor with Skip Reason

@dg.sensor(job=my_job, minimum_interval_seconds=60)
def conditional_sensor(context: dg.SensorEvaluationContext):
    new_files = check_for_new_files()

    if not new_files:
        return dg.SkipReason("No new files found")

    return dg.RunRequest(run_key=f"files_{len(new_files)}")

Partitions

Time-Based Partitions

import dagster as dg

# Daily partitions
daily_partition = dg.DailyPartitionsDefinition(
    start_date="2023-01-01",
    end_date="2024-12-31",  # Optional
)

# Weekly partitions
weekly_partition = dg.WeeklyPartitionsDefinition(
    start_date="2023-01-01",
)

# Monthly partitions
monthly_partition = dg.MonthlyPartitionsDefinition(
    start_date="2023-01-01",
    end_date="2023-12-31",
)

# Hourly partitions
hourly_partition = dg.HourlyPartitionsDefinition(
    start_date="2023-01-01-00:00",
)

Static Partitions

# Fixed set of partitions
region_partition = dg.StaticPartitionsDefinition([
    "us-east",
    "us-west",
    "eu-west",
    "ap-south",
])

# Category partitions
category_partition = dg.StaticPartitionsDefinition([
    "electronics",
    "clothing",
    "home",
    "sports",
])

Dynamic Partitions

Partitions created at runtime based on external data:

from dagster import DynamicPartitionsDefinition

# Define dynamic partition
customer_partition = DynamicPartitionsDefinition(name="customers")

@dg.asset(partitions_def=customer_partition)
def customer_data(context: dg.AssetExecutionContext) -> None:
    customer_id = context.partition_key
    # Process data for this customer
    pass

# Add partitions at runtime
def add_new_customer(customer_id: str):
    from dagster import get_dagster_instance
    instance = get_dagster_instance()
    instance.add_dynamic_partitions("customers", [customer_id])

Use Cases:

• Customer-specific processing where customers are added dynamically
• File-based partitioning where files arrive unpredictably
• Any scenario where partition keys aren't known at definition time

Multi-Dimensional Partitions

multi_partition = dg.MultiPartitionsDefinition({
    "date": dg.DailyPartitionsDefinition(start_date="2023-01-01"),
    "region": dg.StaticPartitionsDefinition(["us", "eu", "ap"]),
})

@dg.asset(partitions_def=multi_partition)
def multi_partitioned_asset(context: dg.AssetExecutionContext) -> None:
    partition_key = context.partition_key
    # partition_key is a MultiPartitionKey with .keys_by_dimension
    date = partition_key.keys_by_dimension["date"]
    region = partition_key.keys_by_dimension["region"]

Using Partitions in Assets

@dg.asset(partitions_def=monthly_partition)
def monthly_data(context: dg.AssetExecutionContext) -> None:
    """Process data for a specific month."""
    partition_date_str = context.partition_key  # "2023-01-01"
    month = partition_date_str[:-3]  # "2023-01"

    context.log.info(f"Processing partition: {month}")

    data = fetch_data_for_month(month)
    save_data(data, month)

Partition Window Access

@dg.asset(partitions_def=daily_partition)
def windowed_data(context: dg.AssetExecutionContext) -> None:
    """Access partition time window."""
    time_window = context.partition_time_window

    start = time_window.start  # datetime
    end = time_window.end      # datetime

    context.log.info(f"Processing from {start} to {end}")

Partitioned Jobs

Define Partitioned Job

partitioned_job = dg.define_asset_job(
    name="partitioned_job",
    selection=["monthly_data"],
    partitions_def=monthly_partition,
)

Run Specific Partition

# In launchpad or programmatically
result = partitioned_job.execute_in_process(
    partition_key="2023-06-01",
)

Backfills

Backfills materialize multiple partitions at once:

UI Backfill

1. Navigate to the asset in the Dagster UI
2. Click "Materialize" dropdown
3. Select "Backfill"
4. Choose partition range

Programmatic Backfill

from dagster import DagsterInstance

instance = DagsterInstance.get()
instance.submit_run(
    partition_key="2023-01-01",
    job_name="partitioned_job",
)

Combining Automation

Partitioned Schedule

# Create job with partition
monthly_job = dg.define_asset_job(
    name="monthly_job",
    selection=["monthly_data"],
    partitions_def=monthly_partition,
)

# Build schedule that runs for new partitions
monthly_schedule = dg.build_schedule_from_partitioned_job(
    job=monthly_job,
)

Sensor with Partitions

@dg.sensor(job=partitioned_job)
def partition_sensor(context: dg.SensorEvaluationContext):
    missing_partitions = get_missing_partitions()

    return [
        dg.RunRequest(
            run_key=f"backfill_{partition}",
            partition_key=partition,
        )
        for partition in missing_partitions
    ]

Anti-Patterns to Avoid

References

• Schedules
• Sensors
• Partitions
• Asset Selection Syntax