Is it possible to build an asset for which we can materialize multiple partitions in a single run without being in a backfill state?

[marcilj]

I have some crypto token prices that I want to store in a table with Dagster.

I'm currently using a MultiPartitionsDefinition defined like this

partitions = MultiPartitionsDefinition(
    {
        "crypto_tokens": DynamicPartitionsDefinition(name="crypto_tokens"),
        "hour": TimeWindowPartitionsDefinition(
            start=datetime(2022, 12, 1),
            fmt="%Y-%m-%d-%H:%M",
            cron_schedule="0 * * * *",
            timezone="Etc/UTC",
            end_offset=1,
        ),
    }
)

This will create one partition for each hours for each tokens.

This is the behavior I want since, it's cleaner in the UI and I can keep track of all the assets states in Dagster and see which ones are missing.

What I want is to be able to schedule the job so it start 1 job for 1 hour for all the crypto_tokens dynamic partitions.

I want that because I can send multiples assets in the body of the query to the API I'm using, and I would be able to reduce the number of calls to the API. Now each for each hour, I'm starting 1 job per crypto_tokens and it's not optimal.

Is there a way I can configured my job to run multiples assets at the time, but still keep the MultiPartitionsDefinition so I can tag which elements was capture and which one was not?

[clairelin135]

Hi @marcilj, thanks for the detail here.

It is possible to kick off a run that targets all dynamic partitions for the given hour, though you'll need to change your dimension names so that the time dimension name is alphabetically before the crypto_token dimension name (this is admittedly awkward behavior, and I explain why this is the case in my comment here).

Once you do this, you can define a schedule that targets a partition range in one run:

@schedule(job=..., cron_schedule="@hourly")
def my_schedule(context):
    ...
    hour_partition_key = ...
    return RunRequest(
        tags={
            "dagster/asset_partition_range_start": MultiPartitionKey(
                {
                    "my_crypto_tokens": dynamic_partitions_def.get_first_partition_key(
                        dynamic_partitions_store=context.instance
                    ),
                    "hour": hour_partition_key,
                }
            ),
            "dagster/asset_partition_range_end": MultiPartitionKey(
                {
                    "my_crypto_tokens": dynamic_partitions_def.get_last_partition_key(
                        dynamic_partitions_store=context.instance
                    ),
                    "hour": hour_partition_key,
                }
            ),
        }
    )

[the4thamigo-uk]

@clairelin135 This is pretty cool, but is it possible to specify a list of partitions rather than a range?