How to define an unpartitioned asset that depends on all upstream partitions

[clairelin135]

For example: I have a daily partitioned asset, and I want to merge all those partitions into a downstream unpartitioned asset which contains all the rows from all partitions.

[clairelin135]

When an unpartitioned asset is downstream of a partitioned asset, dagster automatically infers that the unpartitioned asset depends on all upstream partitions. The opposite applies as well when the unpartitioned asset is upstream of a partitioned asset--each downstream partition depends on the upstream unpartitioned asset.

You can define this relationship in code as follows:

@asset(partitions_def=DailyPartitionsDefinition("2023-01-01"))
def daily_data():
    ...

@asset
def aggregated_report(context, daily_data):
    context.log.info(daily_data) # the default IO manager returns a dictionary of outputs mapped by partition key {"2023-01-01": ..., "2023-01-02": ..., ...}
    ...

Under the hood, dagster applies the AllPartitionMapping to the daily_data input. This defines that all partitions of daily_data are input partitions. Defining this explicitly in code would look like:

@asset(ins={"daily_data": AssetIn(partition_mapping=AllPartitionMapping())})
def aggregated_report(context, daily_data):
    ...

[emirkmo]

This gives an error in Dagster 1.2.7, do you know which version it was fixed?

[clairelin135]

@emirkmo What error are you seeing on 1.2.7? Would it be possible to share a stack trace?

[ChanTheDataExplorer]

How can I create a job or schedule out of it? My scenario is that i have upstream which partitions monthly then downstream asset is used to dedup all the data from the partitions. Im doing it in athena so no passing of data, instead, ctas queries run this pipeline