Ежедневный раздел не может получать и сохранять данные по расписанию. ⇐ Python

[Anonymous]

У меня есть следующий ресурс в дагстере:

Код: Выделить всё

@asset(
partitions_def=DailyPartitionsDefinition(start_date=DATE_INIT_DAILY_ENTSOE),
io_manager_key="postgres_io_manager",
group_name="entsoe_daily",
op_tags={"dagster/concurrency_key": "entsoe", "concurrency_tag": "entsoe"},
description="The day-ahead prices for each market time unit (60m). Day-ahead prices refer to the price per MWh as it is settled each hour in the energy market. In Greece, that is HEnEx.  ENTSOE's data probably comes from there, as IPTO does not list them.",
retry_policy=RetryPolicy(max_retries=3, delay=10)
)
def day_ahead_prices_daily(context: AssetExecutionContext):

try:
partition_time_window = context.partition_time_window
if not partition_time_window:
raise ValueError("Partition time window is not available.")

start, end = timewindow_to_ts(partition_time_window)
context.log.info(f"Handling partition from {start} to {end}")

series: pd.Series = entsoe_client().query_day_ahead_prices(COUNTRY_CODE, start=start, end=end)
dataframe = sanitize_series(series, ['price'])
return Output(value=dataframe)
except requests.exceptions.ConnectionError as e:
context.log.error(f"ConnectionError: {str(e)}")
raise
except NoMatchingDataError:
context.log.warning(f"No matching data for the time window: {start} to {end}")
# Skip the partition by returning a SkipReason
return SkipReason(f"No matching data for the time window: {start} to {end}")
except Exception as e:
context.log.error(f"Unexpected error: {str(e)}")
raise

и я определил следующее задание/график:

Код: Выделить всё

PG_IOMANAGER_CONFIG = {
"username": EnvVar("DATA_PG_USERNAME"),
"password": EnvVar("DATA_PG_PASSWORD"),
"hostname": EnvVar("DATA_PG_HOST"),
"port": EnvVar("DATA_PG_PORT"),
"db_name": EnvVar("DATA_PG_DB")
}

all_assets = load_assets_from_package_module(assets)

entsoe_daily_partitions_def = DailyPartitionsDefinition(start_date=DATE_INIT_DAILY_ENTSOE)

entsoe_daily_job = define_asset_job(
"entsoe_daily_job",
selection=AssetSelection.groups("entsoe_daily"),
partitions_def=entsoe_daily_partitions_def
)

entsoe_daily_schedule = ScheduleDefinition(
job=entsoe_daily_job,
cron_schedule="0 17 * * *",  # daily at 5 PM
)

defs = Definitions(
assets=all_assets,
resources={
"postgres_io_manager": PostgresIOManager(**PG_IOMANAGER_CONFIG)
},
schedules=[
entsoe_daily_schedule
]
)

Однако при автоматическом запуске задания появляется следующая ошибка:

Код: Выделить всё

Exceeded max_retries of 3
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 282, in dagster_event_sequence_for_step
for step_event in check.generator(step_events):
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 494, in core_dagster_event_sequence_for_step
for user_event in _step_output_error_checked_user_event_sequence(
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 183, in _step_output_error_checked_user_event_sequence
for user_event in user_event_sequence:
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 88, in _process_asset_results_to_events
for user_event in user_event_sequence:
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/compute.py", line 198, in execute_core_compute
for step_output in _yield_compute_results(step_context, inputs, compute_fn, compute_context):
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/compute.py", line 167, in _yield_compute_results
for event in iterate_with_context(
File "/usr/local/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 471, in iterate_with_context
with context_fn():
File "/usr/local/lib/python3.10/contextlib.py", line 153, in __exit__
self.gen.throw(typ, value, traceback)
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 72, in op_execution_error_boundary
raise RetryRequestedFromPolicy(
The above exception was caused by the following exception:
dagster._check.functions.CheckError: Failure condition:  Has a PartitionsDefinition, so should either have a partition key or a partition key range
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 54, in op_execution_error_boundary
yield
File "/usr/local/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 473, in iterate_with_context
next_output = next(iterator)
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/compute_generator.py", line 141, in _coerce_op_compute_fn_to_iterator
result = invoke_compute_fn(
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/plan/compute_generator.py", line 129, in invoke_compute_fn
return fn(context, **args_to_pass) if context_arg_provided else fn(**args_to_pass)
File "/opt/dagster/app/datasets/assets/entsoe.py", line 494, in day_ahead_prices_daily
partition_time_window = context.partition_time_window
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/context/compute.py", line 1692, in partition_time_window
return self.op_execution_context.partition_time_window
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/context/compute.py", line 414, in partition_time_window
return self._step_execution_context.partition_time_window
File "/usr/local/lib/python3.10/site-packages/dagster/_core/execution/context/system.py", line 478, in partition_time_window
check.failed(
File "/usr/local/lib/python3.10/site-packages/dagster/_check/functions.py", line 1642, in failed
raise CheckError(f"Failure condition: {desc}")

Подводя итог, я хочу иметь возможность автоматически запускать ресурс в соответствии с расписанием cronjob. Я заметил, что могу запустить его вручную из пользовательского интерфейса дагстера, но когда он запускается автоматически, я получаю ошибку. Есть идеи, как я могу это решить?

Подробнее здесь: https://stackoverflow.com/questions/787 ... e-schedule