Is it possible to dynamically set dagster/concurrency_key from partitions at run time?

[chrishiste]

I have a job that runs against multiple Cloud regions, each backed by a separate data store. I want to enforce a concurrency limit such that, for each region, I never exceed 10 queries running at once in a specific @op.

However, I don’t want one region’s concurrency limit to affect another’s—each region should be governed by its own limit.

What I’m trying to achieve:
I’d like to dynamically set the dagster/concurrency_key for each run based on a partition value.

My partitions are dynamic (dagster.DynamicPartitionsDefinition), and I can’t know or hardcode all the possible dagster/concurrency_key values at definition time.

Is there a supported way to achieve this behavior?

[C-Loftus]

Curious, did you find a way to address this since you first posted?

I am also trying to figure this out and it seems like implementing the logic in Python is too disconnected from the rest of dagster but there aren't other ways of doing it with the tag/metadata system dynamically

[chrishiste]

Unfortunately I ended up having to write a custom sensor that manages my own scheduling and parallelization. I’m essentially emitting RunRequests directly and I’ve stopped using auto-materialization/schedules altogether.

It’s gotten really complex, and I also hit other Dagster limitations around sensors. At this point I’m basically abusing the sensor system to make it work because of Dagster limitations.

To be honest, on paper Dagster is great for partitioned jobs, but in practice it still lacks a lot of basic features you end up having to handle yourself.

[C-Loftus]

That makes sense. Thank you for your response.

[william-pearsall-foodsmart]

I was in the same boat, but at least currently this is trivial to do with unique tags in the queued run coordinator.

https://docs.dagster.io/guides/operate/managing-concurrency#limit-the-number-of-runs-that-can-be-in-progress-by-unique-tag-value

concurrency:
  runs:
    tag_concurrency_limits:
      - key: 'partition/YOURDYNAMICPARITION'
        value:
          applyLimitPerUniqueValue: true
        limit: 1

I'm currently on dagster 1.12, but it appears this goes back at least to 1.10.

[C-Loftus]

Thanks for your reply!

I think this can definitely be a good solution for some folks, but unfortunately I don't there is a way to set different limits for different subsets of the partition, beyond giving all keys in the entire partition the same limit. Let me know if I am wrong or missing something though.

i.e. here is a contrived pseudo-yaml example of an ideal concurrency limit

shopping_data_partition (contains list of all shops which is dynamically generated)
    - shops in region A
        - limit: 4
    - shops in region B
        - limit: 8  

Since the list of shops is dynamically generated you can't set a tag.

The only way I think you could do this is to separate the partition into two disjoint partitions (i.e. one for all of region A and another for all of region B. However, then it makes harder to keep all shops in sync and starts to decouple them when there isn't a semantic reason to do so)

[chrishiste]

Yeah for this to work you would need to know in advance all the dynamic values but might work for some :)