https://dagster.io/ logo
#ask-community
Title
# ask-community
a

Alec Ryan

05/28/2022, 6:46 PM
How can I define a daily partition in one place that gets used in many different assets?
dagster bot resolve to issue 1
o

owen

05/31/2022, 10:10 PM
hi @Alec Ryan! Are you looking for something different than
Copy code
my_partitions_def = DailyPartitionsDefinition(start_date="2022-01-01")
and passing that into the
partitions_def
argument of the asset decorator?
a

Alec Ryan

05/31/2022, 10:16 PM
Yeah, I'm looking for a way to pass this to job or asset group if possible?
o

owen

05/31/2022, 11:08 PM
ah i see what you mean -- this is not possible at the moment unfortunately
I'll create an issue for this ability, as it does seem useful to "bulk apply" a partitions def to a bunch of assets at once
@Dagster Bot issue API for applying a PartitionsDefinition to many assets at once
d

Dagster Bot

05/31/2022, 11:09 PM
a

Alec Ryan

05/31/2022, 11:40 PM
Thanks @owen
How would I be able to do something similar with ops? Right now I'm doing this which seems a bit hacky:
Copy code
from dagster import daily_partitioned_config
from datetime import datetime

@daily_partitioned_config(start_date=datetime(2020, 1, 1))
def nhl_elt_daily_config(start: datetime, _end: datetime):
    return {
        "ops": {
            "extract_game_ids_to_list": {"config": {"date": start.strftime("%Y-%m-%d")}},
            "load_game_data_to_s3": {"config": {"date": start.strftime("%Y-%m-%d")}},
            "delete_partition_from_snowflake": {"config": {"date": start.strftime("%Y-%m-%d")}},
            "copy_partition_into_snowflake": {"config": {"date": start.strftime("%Y-%m-%d")}},
            "dbt_run": {"config": {"run_date": start.strftime("%Y-%m-%d")}},
            "dbt_test": {"config": {"run_date": start.strftime("%Y-%m-%d")}}
        }
    }
I basically want to define run_date once and pass that to all of the ops as if it were a resource
o

owen

06/01/2022, 4:12 PM
I would actually recommend doing exactly that -- you can create a resource called "run_date", with something like
resource_defs={"run_date": ResourceDefinition.string_resource()}
, then when you're returning config, have a block for
{..., "resources": {"run_date": {"config": start.strftime(...)}}}
from there, you can just reference
context.resources.run_date
inside the ops instead of context.op_config
a

Alec Ryan

06/01/2022, 4:15 PM
And that will work with the partitioned config?
in the partition def I can specify the resource and then use that resource in ops
o

owen

06/01/2022, 4:23 PM
yeah exactly, so instead of returning all the op config inside the nhl_elt_daily_config, you can return just the single bit of resource config and then use the resource in your ops
a

Alec Ryan

06/01/2022, 4:23 PM
Awesome. Thanks for the info I'll give that a shot
I want to specify that resource in my job:
Copy code
@job(
    config=nhl_elt_daily_config,
    resource_defs={
        "s3": s3_resource,
        "snowflake": snowflake_resource_configured,
        "dbt": dbt_resource_configured
        }
    )
like this ^
So then in my op do I specify that resource like this?
Copy code
@op(
    required_resource_keys={'dbt', 'run_date'}, 
    ins={"start": In(Nothing)}
    )
def dbt_run(context):
    context.resources.dbt.run()
config:
Copy code
from dagster import daily_partitioned_config
from datetime import datetime

@daily_partitioned_config(start_date=datetime(2020, 1, 1))
def nhl_elt_daily_config(start: datetime, _end: datetime):
    return {
        "resource": {
            "run_date": {
                "config": start.strftime("%Y-%m-%d")
            }
        }
    }
Where I am confused is how I can pass this into the job. Not sure if this should be in the config or the resource def
o

owen

06/01/2022, 5:02 PM
gotcha yeah -- so the full thing based off of this setup would be
Copy code
@job(
    config=nhl_elt_daily_config,
    resource_defs={
        "s3": s3_resource,
        "snowflake": snowflake_resource_configured,
        "dbt": dbt_resource_configured,
        "run_date" ResourceDefinition.string_resource(),
        }
    )
then
Copy code
@op(
    required_resource_keys={'dbt', 'run_date'}, 
    ins={"start": In(Nothing)}
    )
def dbt_run(context):
    my_run_date = context.resources.run_date
    context.resources.dbt.run()
then
Copy code
from dagster import daily_partitioned_config
from datetime import datetime

@daily_partitioned_config(start_date=datetime(2020, 1, 1))
def nhl_elt_daily_config(start: datetime, _end: datetime):
    return {
        "resources": {
            "run_date": {
                "config": start.strftime("%Y-%m-%d")
            }
        }
    }
basically, it should be part of both. inside of resource_defs, you're saying that your job will have some resource called "run_date" of type "string resource". then, when you're configuring your job, you'll need to provide config for that resource. you don't need to do this for your other resources because (it seems) you've already called
.configured()
on them (so all their config is already satisfied)
a

Alec Ryan

06/01/2022, 5:04 PM
got it and this is a fairly common design pattern?
and correct I have a separate directory for resources
o

owen

06/01/2022, 5:06 PM
yep! we have an example that uses this pattern (the hacker_news example). this job: https://sourcegraph.com/github.com/dagster-io/dagster/-/blob/examples/hacker_news/hacker_news/jobs/hacker_news_api_download.py has a "partition_bounds" resource that specifies a start/end time, which a bunch of its ops use
then, there's an hourly_partitioned_config function that will provide config for that resource based off of the partition: https://sourcegraph.com/github.com/dagster-io/dagster/-/blob/examples/hacker_news/hacker_news/schedules/hourly_hn_download_schedule.py
a

Alec Ryan

06/01/2022, 5:06 PM
I was loosely following that project which is how I ended up here lol
o

owen

06/01/2022, 5:06 PM
haha
a

Alec Ryan

06/01/2022, 5:07 PM
Thanks again!
Oh one more thing
I use that run_date as a var in dbt typically
Is there an easy way to pass that to a dbt run operator?
o

owen

06/01/2022, 5:11 PM
yup!
context.resources.dbt.run(vars={"my_dbt_var": my_string})
a

Alec Ryan

06/01/2022, 5:12 PM
beautiful
you guys thought of everything lol
o

owen

06/01/2022, 5:12 PM
the dbt integration is my baby haha -- glad it's working for you 😄
30 Views