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

Nicholas Cohn-Martin

07/12/2022, 6:14 PM
Hi Team, when passing a run_request_for_partition are you able to include a run_config? When trying to pass it I get
TypeError: run_request_for_partition() got an unexpected keyword argument 'run_config'
y

Yang

07/12/2022, 6:40 PM
Hey team, we need the sensor to pass op configs basically
y

yuhan

07/12/2022, 7:43 PM
You can specify the op config on the job itself.
Copy code
def partition_fn(partition_key: str):
    return {"ops": {"my_op": {"config": {"partition": partition_key}}}} # here


@static_partitioned_config(partition_keys=["a", "b", "c", "d"])
def my_partitioned_config(partition_key: str):
    return partition_fn(partition_key)


@op
def my_op(context):
    print(context.op_config["partition"])


@job(config=my_partitioned_config)
def my_job():
    my_op()


@schedule(cron_schedule="* * * * *", job=my_job, execution_timezone="US/Pacific")
def my_schedule(_context):
    yield my_job.run_request_for_partition(partition_key="a", run_key=None)


@repository
def my_repo():
    return [my_job, my_schedule]
s

sandy

07/12/2022, 8:14 PM
@yuhan FYI I think they're using an asset job
y

Yang

07/12/2022, 8:15 PM
we'd like to pass some vars somehow from the sensor to the job/op
Copy code
"ops": {
                                "unzip_and_write_to_gcs_date": {
                                    "config": {
                                        "zipfilepath": full_file_name ,
                                        "xmlfolder": run_key}}}
s

sandy

07/12/2022, 8:23 PM
this isn't currently possible, but I think it might be possible for us to add it in our next release. taking a look
y

Yang

07/12/2022, 8:24 PM
oh shoot, when's the next release?
s

sandy

07/12/2022, 8:24 PM
thursday
y

Yang

07/12/2022, 8:24 PM
oh ok
s

sandy

07/12/2022, 8:26 PM
would it be possible for your code to figure out full_file_name using the partition key? e.g. if you're launching a run for
2022-12-12
, the file name might be something like "zipfiles/20221212"
y

Yang

07/12/2022, 8:31 PM
yeah I think we can change it around
Oh wait, do we have access in ops to the run_key? We have it partitioned on date
Do we need to change the setup to be partitioned by file?
we just a different partitions definition?
Copy code
daily_funds_job = define_asset_job(
    "daily_funds_job",
    selection=AssetSelection.assets(fund_holdings_asset),
    partitions_def=daily_partitions_def)
Oh I see, no we can't get the filename just from the date. We really need to pass it bc there are a few variations.
We can pass op configs if the job is not partitioned, though, right? Looks like RunRequest can get a
run_config
s

sandy

07/12/2022, 10:05 PM
Ah, got it. And yes, you can pass op configs if the job is not partitioned
y

Yang

07/12/2022, 10:11 PM
ok!
s

sandy

07/12/2022, 10:28 PM
Would working without partitions work for you in this case? I think maybe a core issue here is that partitions in Dagster are tied closely with backfills - i.e. being able to go to an asset or job and kick off something that rematerializes all the partitions in history from scratch. But if the filenames aren't deterministic from the partition keys, then that backfill wouldn't be able to run
y

Yang

07/12/2022, 10:37 PM
Yeah, I think we don't really need partitions