With Dagster deployed on Kubernetes I am trying to materiali dagster #ask-community

With Dagster deployed on Kubernetes, I am trying t...

Aaron T

03/31/2023, 5:19 PM

With Dagster deployed on Kubernetes, I am trying to materialize a downstream TimeWindowPartition and I am getting this error, even though I know that the upstream assets exist (currently in s3) - Any idea what might cause this error? I was thinking it could be the IO manager, but not sure

Copy code

dagster._check.CheckError: Failure condition: Tried to access partition key for asset 'AssetKey(['alarm_watchdog_report_graph'])', but the number of input partitions != 1: 'TimeWindowPartitionsSubset([PartitionKeyRange(start='2023-03-27', end='2023-03-28')])'

dagster bot responded by community 1

🤖 1

Aaron T

03/31/2023, 5:30 PM

Using the

dagster_aws/s3/io_manager.py

So, now I am thinking the dagster_aws IO manager does not currently support getting TimeWindowPartionsSubset

Timo Vink

03/31/2023, 5:31 PM

I have also been struggling with this. In my case it's a dynamic partition:

Copy code

partitions_def = DynamicPartitionsDefinition(name='top_stories')

@asset(partitions_def=partitions_def)
def top_stories(context):
    # omitted

@asset
def unique_top_stories(top_stories):
    # omitted

If I have exactly 1 partition I get that partition's value in

unique_top_stories

(was expecting some collection). If I have multiple partitions I get the same error as Aaron. I tried explicitly specifying an

AssetIn

and

AllPartitionMapping

but no change.

Timo Vink

03/31/2023, 5:41 PM

I think you're right about this being tied to the IO manager @Aaron T. Confirmed just now that if I use the default IO manager my assets work, but switching to the following (pointing at empty bucket) breaks.

Copy code

s3_pickle_io_manager.configured({ 's3_bucket': 'my_bucket_here', 's3_prefix': 'my_prefix_here'         })

Timo Vink

03/31/2023, 5:54 PM

Brand new to

dagster

and its code base but looking at the source code comparing the FS IO manager and S3 IO manager I see these inheritance chains:

PickledObjectFilesystemIOManager

UPathIOManager

MemoizableIOManager

-> ... and

PickledObjectS3IOManager

MemoizableIOManager

-> ...

UPathIOManager

specifically says "handles partitioned assets", and comparing the following two methods seems to support the idea that the S3 IO manager just doesn't handle partitions. https://github.com/dagster-io/dagster/blob/8a83cd5d52e5be18dd0ab7a991cf48f594feedbd/python_modules/dagster/dagster/_core/storage/upath_io_manager.py#L187 https://github.com/dagster-io/dagster/blob/8a83cd5d52e5be18dd0ab7a991cf48f594feedb[…]thon_modules/libraries/dagster-aws/dagster_aws/s3/io_manager.py

Timo Vink

03/31/2023, 5:55 PM

This issue seems relevant: https://github.com/dagster-io/dagster/issues/10908 But crucially is missing the actual exception message, which is why I missed it before.

Timo Vink

03/31/2023, 7:32 PM

@Aaron T, in case this is useful to you: https://github.com/dagster-io/dagster/issues/13290 It includes a custom S3 IO manager that works for me.

Aaron T

03/31/2023, 7:33 PM

thanks for the heads up! I will take a look

3 Views

Open in Slack

Previous Next