Is it possible to specify that a partitioned asset should execute the materialization of the individual partitions sequentially?

Yes, it is possible to specify that a partitioned asset should execute the materialization of individual partitions sequentially. The documentation suggests that one way to enforce sequential execution order for asset backfills and auto-materialization is to define a partition mapping where each partition depends on the prior partition. This approach would make Dagster respect the partition ordering. However, it's important to note that there isn't a built-in way to enforce this ordering directly, and workarounds like defining a partition mapping are suggested. For example, a user in a discussion mentioned that reducing the concurrency limit to 1 did not help to fix the execution order, and the suggested workaround was to define a partition mapping with dependencies between partitions. If you need to enforce a specific execution order for your use case and the suggested workaround does not meet your needs, you may consider filing a feature request as per the guidance provided in the discussions.

@Scout How can I specify a partition mapping where each partition depends on the prior partition?

To specify a partition mapping where each partition depends on the prior partition, you can use a custom

PartitionMapping

. A

PartitionMapping

defines the relationship between partitions of different assets or between different partitions of the same asset. In the case of sequential dependencies, where each partition should depend on the previous one, you would need to create a custom

PartitionMapping

that expresses this relationship. The documentation does not provide a direct example of such a custom

PartitionMapping

, but it does mention the concept of overriding default partition dependency rules by providing a

PartitionMapping

. Here's a conceptual example based on the documentation's guidance on partition mappings:

python
from dagster import PartitionMapping, PartitionKeyRange

class SequentialPartitionMapping(PartitionMapping):
    def get_upstream_partitions_for_partition_range(
        self,
        downstream_partition_key_range: PartitionKeyRange,
        downstream_partitions_def,
        upstream_partitions_def,
    ) -> PartitionKeyRange:
        # Assuming downstream_partition_key_range has a single partition key
        downstream_partition_key = downstream_partition_key_range.start
        # Calculate the previous partition key based on your partition scheme
        previous_partition_key = calculate_previous_partition_key(downstream_partition_key)
        return PartitionKeyRange(previous_partition_key, previous_partition_key)

    def get_downstream_partitions_for_partition_range(
        self,
        upstream_partition_key_range: PartitionKeyRange,
        downstream_partitions_def,
        upstream_partitions_def,
    ) -> PartitionKeyRange:
        # This method would define which downstream partitions are affected by an upstream range
        # For sequential execution, you might not need to implement this method
        raise NotImplementedError()

# Use the custom partition mapping in your asset definition
@asset(partitions_def=my_partitions_def, ins={"my_input": AssetIn(partition_mapping=SequentialPartitionMapping())})
def my_asset(context, my_input):
    # Asset computation logic here
    pass

In this example,

calculate_previous_partition_key

is a placeholder for a function you would write to determine the previous partition key based on your partitioning scheme. Please note that this is a conceptual example to illustrate the approach. The actual implementation will depend on your specific partitioning scheme and how you define and calculate partition keys. You may need to adjust the logic to fit the requirements of your partitions. If you need further guidance on implementing a custom

PartitionMapping

for your specific use case, you can refer to the Dagster documentation or seek help from the Dagst