How to: execute daily-partitioned job after a file arrives or other external condition is met

[sryza]

Data arrives from an external source every day, but it's hard to make guarantees about the time it arrives. We'd like to run a job to process the data (and likely materialize some asset partitions) as soon as the data arrives.

[sryza]

Here's an implementation of a sensor that checks for some external condition to be true (like a file arriving in a filesystem) for a particular date.

Note that, if a file never arrives for a particular date, this sensor won't move past that date. This is desirable for some applications, but undesirable for other applications. It's also possible to write the sensor in a way that it will move past dates.

from dagster import (
    sensor,
    asset,
    SkipReason,
    repository,
    JobDefinition,
)
from datetime import datetime, timedelta


DATE_FORMAT = "%Y-%m-%d"
START_DATE = "2022-03-01"


def file_exists_for_date(date: str) -> bool:
    return ...


partitioned_job = ...
assets = ...

# Define the sensor
@sensor(job=partitioned_job)
def date_file_sensor(context):
    last_processed_date = context.cursor
    if last_processed_date is None:
        next_date = START_DATE
    else:
        next_date = (
            datetime.strptime(last_processed_date, DATE_FORMAT) + timedelta(days=1)
        ).strftime(DATE_FORMAT)

    if file_exists_for_date(next_date):
        context.update_cursor(next_date)
        return partitioned_job.run_request_for_partition(next_date, run_key=None)
    else:
        return SkipReason(f"Did not find file for next expected date {next_date}")


@repository
def my_repo():
    return [partitioned_job, date_file_sensor, assets]

partitioned_job can either be defined with ops or with software-defined assets. Here's an example of how you might define it using software-defined assets:

from dagster import AssetSelection, DailyPartitionsDefinition, asset, define_asset_job

daily_partitions_def = DailyPartitionsDefinition(start_date=START_DATE)

@asset(partitions_def=daily_partitions_def)
def my_daily_asset(context):
    date = context.partition_key
    contents = get_file_contents_for_date(date)
    get_dagster_logger().info(f"Processing file for date {date}")


@asset(partitions_def=daily_partitions_def)
def my_downstream_daily_asset(context, my_daily_asset):
    ...

assets = [my_daily_asset, my_downstream_daily_asset]
partitioned_job = define_asset_job("partitioned_job", selection=AssetSelection.assets(*assets), partitions_def=daily_partitions_def)