Heya can someone please tell me what I m doing wrong I set u dagster #ask-community

Heya, can someone please tell me what I'm doing wr...

Matthew Heguy

03/22/2023, 5:39 PM

Heya, can someone please tell me what I'm doing wrong? I set up a bunch of schedules and I thought everything was fine, but they're all generating invalid configurations.. I've reduced the problem down to just a few statements in the attached py file and I included a screenshot of the code for quick reference.

iso_test.py

jamie

03/22/2023, 8:31 PM

hey @Matthew Heguy could you copy paste the python file in the thread?

Matthew Heguy

03/23/2023, 12:53 PM

Copy code

from datetime import datetime

from dagster import (
    InitResourceContext,
    ScheduleDefinition,
    ScheduleEvaluationContext,
    job,
    op,
    resource,
    validate_run_config,
)


@resource(config_schema={"environment": str})
def api_client(context: InitResourceContext) -> str:
    return f'Resource: {context.resource_config["environment"]}'


@op(required_resource_keys={"api_client"})
def my_op(context) -> None:
    api_client_res = context.resources.api_client
    print(api_client_res)


@job(resource_defs={"api_client": api_client})
def my_job() -> None:
    my_op()


my_schedule = ScheduleDefinition(
    job=my_job,
    cron_schedule="* * * * *",
    run_config=validate_run_config(
        my_job,
        {"resources": {"api_client": {"config": {"environment": "prod"}}}},
    ),
)
# Running the job with the config works
my_job.execute_in_process(
    run_config={"resources": {"api_client": {"config": {"environment": "prod"}}}}
)
# Running the schedule produces a bad run config
result = my_schedule.evaluate_tick(
    context=ScheduleEvaluationContext(None, datetime(2000, 1, 1, 0, 0, 0))
)
run_config = result.run_requests[0].run_config
print(run_config)
validate_run_config(my_job, run_config)  # Throws exception

ty spinny 1

jamie

03/23/2023, 1:47 PM

ok i think the issue for you is that

validate_run_config

returns full configuration for the entire run of the job. This will include configuration information for how to execute the run (ie multiprocess, single process, on kubernetes, etc), logging, and all kinds of other stuff. However, the configuration just needed for

my_job

{"resources": {"api_client": {"config": {"environment": "prod"}}}}

so when you run

validate_run_config

in your schedule definition, you’re telling dagster to use the full configuration (including logging info, execution info etc) as the configuration for

my_job

instead you should do this

Copy code

my_schedule = ScheduleDefinition(
    job=my_job,
    cron_schedule="* * * * *",
    run_config={"resources": {"api_client": {"config": {"environment": "prod"}}}},
)

Matthew Heguy

03/23/2023, 1:50 PM

You're quite right, thank you very much!

👍 1

2 Views

Open in Slack

Previous Next