How to schedule an hourly job on a daily partitioned asset

[yusufshalaby]

I have daily asset partitions and would like to refresh the current day's partition every hour. How can this be done without changing my asset partition scheme?

[sryza]

Hey @yusufshalaby - something like this should work:

from dagster import DailyPartitionsDefinition

daily_partitions_def = DailyPartitionsDefinition(start_date=..., end_offset=1)

@asset(partitions_def=daily_partitions_def)
def asset1():
    ...

@schedule(cron_schedule="@hourly", job=define_asset_job("asset1_job", selection=[asset1]))
def schedule1(context):
    partition_date = context.scheduled_execution_time.replace(hour=0)
    partition_key = partition_date.strftime(daily_partitions_def.fmt)
    return RunRequest(partition_key=partition_key)

Note the use of end_offset=1. This allows there to be a partition for the current day. Without that, there's only partitions for days that have completed.

[adityawarmanfw]

Hi @sryza, how can I get the execution_time if I use sensor instead of schedule? So I want to materialize daily partitioned assets, which depends on non partitioned upstreams that are scheduled hourly. Thanks

[sryza]

@adityawarmanfw - sensors don't have a notion of scheduled_execution_time.

What exact scheduling logic are you looking to implement? I.e. do you want to materialize the daily asset every time that the upstream asset gets updated?

[yusufshalaby]

Thanks! Slightly out of scope of the original question, but is there a way to pass a partition key range instead of just a partition key? I basically want a schedule with nightly refreshes of all partitions and hourly refreshes of the current day's partition.

I can accomplish this right now with run_config_fn but it would be nice to incorporate the partitioning functionality.

[sryza]

Here's an example of how to do this: #14622

[yusufshalaby]

Thank you!

[jackson-burke]

I know this thread hasn't been active for awhile, but I'm having trouble with the suggested implementation, specifically at midnight.

Ideally, at midnight on 2024-06-18, my partition.start = 2024-06-17 00:00 and my partition.end = 2024-06-18 00:00 but in reality, I end up with both start and end equal to 2024-06-18 00:00, and therefore, I don't pull any data since in my API call, I give a start and end time.

Is there a way to handle this problem where I'd like the full partition window to be 24 hours but instead it's showing at 0 hours?

Any help would be appreciated!