Nicholas Pezolano
07/10/2023, 4:16 PM@asset(partitions_def=partition_def)
def my_asset(context):
current_partition_window = context.partition_time_window
prev_window = daily_partitions_def.get_prev_partition_window(current_partition_window.start)
prev_partition_key = prev_window.start
return prev_partition_key
my_job = define_asset_job("my_job", selection=[my_asset], partitions_def=partitions_def)
e.g. if my partition is:
partitions_def = DailyPartitionsDefinition(start_date='2023-07-03', timezone='America/New_York', fmt='%Y-%m-%d', end_offset=1)
and my scheduled job is:
def skip_july_4th(context) -> bool:
dt = context.scheduled_execution_time
if '2023-07-04' in str(dt):
return False
else:
return True
@schedule(job=my_job, should_execute=skip_july_4th, cron_schedule="40 22 * * 1-5", execution_timezone='America/New_York')
def nyse_schedule(context):
partition_key = daily_partitions_def.get_partition_key_for_timestamp(
context.scheduled_execution_time.timestamp()
)
request = us_job.run_request_for_partition(partition_key=partition_key)
yield request
The example above will return a previous partition key of '2023-07-04' on '2023-07-05' even thou it will return False for the given should_execute
function in the schedule. How can I get the previous scheduled partition key in the asset?chris
07/10/2023, 4:56 PMget_latest_materialization_event
, but that doesn’t necessarily work if, for example, you try to backfill. I think you need to retrieve the asset materialization event with the highest partition, and I’m not sure if there’s an easy way to do that. Surfacing this to a discussionchris
07/10/2023, 4:57 PMDagster Bot
07/10/2023, 4:57 PMNicholas Pezolano
07/10/2023, 5:04 PMchris
07/10/2023, 5:04 PMNicholas Pezolano
07/10/2023, 5:05 PMNicholas Pezolano
07/10/2023, 5:09 PMNicholas Pezolano
07/10/2023, 5:19 PMchris
07/10/2023, 5:25 PMchris
07/10/2023, 5:25 PMNicholas Pezolano
07/10/2023, 5:26 PMchris
07/10/2023, 5:46 PMshould_execute
might be more robust in that regard