Непонятная ошибка Dagster. Состояние сбоя: попытка доступа к assets_partitions_time_window, но актив не секционирован. ⇐ Python

[Anonymous]

Я создал DeltaIOManager, в котором метод load_input имеет метаданные filter_datetime_on:

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

if context.has_partition_key and filter_datetime_on:
context.log.info(f"Loading partition key: {context.partition_key}")
start, end = context.asset_partitions_time_window
start = start - timedelta(days=1)
end = end + timedelta(days=1)
context.log.info(f"Filtering time {start}-{end}")

идея состоит в том, чтобы уже фильтровать вводимые данные, чтобы сохранить в памяти данные для актива с большим количеством входных данных. хотя это работает для некоторых входных данных, для некоторых это не удалось. самое странное, что все входные данные определены одинаково:

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

"df": AssetIn(
input_manager_key="delta",
metadata={"filter_datetime_on": "time"},
key=AssetKey(["silver", "schema", "df"]),
),

и самое СТРАННОЕ, когда это не удалось, ошибка:

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

dagster._check.CheckError: Failure condition: Tried to access asset_partitions_time_window, but the asset is not partitioned.

Однако ошибка возникла после загрузки раздела_ключа, поэтому context.has_partition_key=True. Как может возникнуть ошибка, если context.has_partition_key=True, а ресурс имеет part_definition?

Подробнее здесь: https://stackoverflow.com/questions/789 ... tions-time