https://dagster.io/ logo
Title
r

Remy Dufrenoy

05/11/2021, 8:04 AM
Hello everyone! I am trying to implement an example given here https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules but I am having troubles making it work. The specific example on the page is this one:
@daily_schedule(
    start_date=datetime.datetime(2020, 1, 1),
    pipeline_name="my_pipeline",
    solid_selection=preset.solid_selection,
    mode=preset.mode,
    tags_fn_for_date=lambda _: preset.tags,
)
def my_modified_preset_schedule(date):
    modified_run_config = preset.run_config.copy()
    modified_run_config["date"] = date.strftime("%Y-%m-%d")
    return modified_run_config
When I am selecting the partition in dagit's playground, the config is updated and the right date is picked up. However, when I try to run a backfill, the config is not updated for each run. Instead, it seems to take only the config of the last partition selected in dagit. Here is the exact code for the schedule I am trying to run, I created it in isolation to make sure it wasn't something else in my code causing the issue:
@solid(config_schema={"test_in": str})
def backfill_solid(context):
    test_in = context.solid_config["test_in"]
    <http://context.log.info|context.log.info>(f"test_in is {test_in}")



MODE_BUG = ModeDefinition(name="MODE_BACKFILL_BUG", resource_defs={"io_manager": fs_io_manager})

PRESET_BACKFILL_BUG = PresetDefinition.from_files(
    name="PRESET_BACKFILL_BUG",
    mode="MODE_BACKFILL_BUG",
    config_files=[
        file_relative_path(__file__, "./presets_backfill_bug.yaml")
    ])



@pipeline(mode_defs=[MODE_BUG],
          preset_defs=[PRESET_BACKFILL_BUG])
def backfill_pipeline():
    backfill_solid()



@daily_schedule(
    pipeline_name="backfill_pipeline",
    start_date=datetime.datetime(2021, 5, 1),
    execution_time=datetime.time(9, 0, 0),
    solid_selection=PRESET_BACKFILL_BUG.solid_selection,
    mode=PRESET_BACKFILL_BUG.mode,
)
def backfill_schedule(date):
    modified_run_config = PRESET_BACKFILL_BUG.run_config.copy()
    modified_run_config["solids"]["backfill_solid"]["config"]["test_in"] = date.strftime("%Y-%m-%d")
    return modified_run_config
I attached the run configuration I see in dagit for one of the runs I use Dagster 0.11.8 and my Python version is 3.8.5
d

daniel

05/11/2021, 1:20 PM
Hi Remy - just a hunch here, does changing it to
PRESET_BACKFILL_BUG.run_config.deepcopy()
(deepcopy instead of copy) fix the problem?
(If so, that probably indicates something that we should fix on our end / that we should modify the example you based the code from)
r

Remy Dufrenoy

05/11/2021, 2:07 PM
Hi daniel, I tried your suggested change but I can't call deepcopy() on PRESET_BACKFILL_BUG.run_config However, this one works
import copy
...
modified_run_config = copy.deepcopy(PRESET_BACKFILL_BUG.run_config)
...
Thanks for you help!
:condagster: 1
d

daniel

05/11/2021, 6:37 PM
@Dagster Bot docs schedule from preset example should use deepcopy
d

Dagster Bot

05/11/2021, 6:37 PM