Dagster通过PartitionSetDefinition创建分区时如何提供预设配置

PartitionSetDefinition 的设计本身不依赖 Pipeline 预设完成配置绑定，所有对应模式下的环境配置、资源配置，都可以通过其内置的 run_config_fn_for_partition 回调函数动态注入，这也是官方推荐的标准实现路径，无需额外适配预设挂载能力。

具体实现方式 #

1. 直接在配置回调中组装模式对应的固定配置 #

你可以提前将不同模式下的公共配置、资源定义抽成独立的可复用配置块，在生成分区运行配置时，将模式基础配置与单个分区的专属配置深度合并后返回：

from dagster.utils import deep_merge_dicts

# 提前定义不同模式下的基础配置（结构和yaml定义完全一致）
MODE_BASE_CONFIG = {
    "test": {
        "resources": {
            "storage": {
                "config": {"root_path": "/tmp/local_dev_storage"}
            }
        }
    },
    "prod": {
        "resources": {
            "storage": {
                "config": {"root_path": "s3://production-bucket/dagster"}
            }
        }
    }
}

# 改造分区配置生成函数，支持配置合并
def run_config_for_date_partition(partition):
    # 生成分区专属配置，比如按分区值传入日期参数
    partition_config = {
        "ops": {
            "filter_daily_data": {
                "config": {"date": partition.value.isoformat()}
            }
        }
    }
    # 取当前分区集绑定模式的基础配置，和分区配置做深合并
    base_config = MODE_BASE_CONFIG["test"]
    return deep_merge_dicts(base_config, partition_config)

注意：PartitionSetDefinition 传入的 mode 参数会在运行创建时自动切换执行上下文的资源集，你只需要在返回的 run_config 中填好对应模式下的资源、算子配置即可，不需要额外手动指定 mode 字段。

2. 复用 Pipeline 已有预设的配置值 #

如果你不想重复维护一套配置字典，完全可以直接从 Pipeline 已经定义好的预设中提取配置值，注入到分区配置回调中，避免配置冗余：

def run_config_for_date_partition(partition):
    # 直接从Pipeline中读取指定预设的配置值
    preset_base_config = my_pipeline.get_preset("test").run_config
    partition_config = {
        "ops": {
            "filter_daily_data": {
                "config": {"date": partition.value.isoformat()}
            }
        }
    }
    return deep_merge_dicts(preset_base_config, partition_config)

3. 直接加载外部Yaml配置文件 #

如果你的模式配置是存储在独立的环境yaml文件中，可以在配置回调中直接加载yaml内容作为基础配置，完全保留原有的yaml配置维护习惯：

from dagster import load_yaml_from_path

def run_config_for_date_partition(partition):
    # 直接加载对应模式的yaml配置文件
    base_config = load_yaml_from_path("config/test_environment.yaml")
    partition_config = {
        "ops": {
            "filter_daily_data": {
                "config": {"date": partition.value.isoformat()}
            }
        }
    }
    return deep_merge_dicts(base_config, partition_config)

常见注意事项 #

• 不要尝试给 PartitionSetDefinition 传入预设相关参数，该类本身没有预设绑定的设计入口：预设是面向手动运行场景的快捷配置入口，分区运行的配置天然需要结合分区上下文动态生成，必须通过 run_config_fn_for_partition 实现。
• 多份配置合并时必须使用深合并逻辑，直接用 Dagster 内置的 deep_merge_dicts 工具即可，不要自行实现浅拷贝逻辑，避免模式级别的资源配置被分区配置意外覆盖。
• 同一套代码如果需要支持多模式的分区集，只需要将 mode 值、对应基础配置块抽成可传入的变量即可，无需依赖额外的预设挂载能力。

内容的提问来源于stack exchange，提问作者j-hulbert