19913d85-f662-4518-a9a1-7adc1793dfd4.png){kind=small}
Channels
#ask-community
Title
# ask-community
l
Lindsay S
12/20/2022, 10:41 PMi'm trying to follow the docs for creating a partition asset job (and eventually want to schedule partitioned assets to materialize daily). My asset works and I can manually materialize and backfill assets, but when I try to add a job (so that I can eventually add a schedule) the code deployment fails `dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code serv`er. I keep going over the docs and am not sure where I am going wrong. If someone could point me to a good example or help me 1. understand if this is the correct way to add schedules for materializing partitioned assets daily - the plan is to get the job working and schedule the job using a scheduler 2. help me understand why I am unable to load this job. Please see code below
🤖 1
Copy code
daily_partitions_def = DailyPartitionsDefinition(start_date="2022-01-19")
@asset(
group_name="my_table_asset",
partitions_def=daily_partitions_def,
required_resource_keys={"some_api_client"}
)
def my_table_asset(context):
date = context.asset_partition_key_for_output()
context.resources.some_api_client.fetch_data(from_date_str=date)
...
partitioned_asset_job = define_asset_job(
name="my_table_asset_job",
selection=AssetSelection.assets(my_table_asset),
partitions_def=daily_partitions_def,
)
@repository
def extract_load():
return [
partitioned_asset_job, # when I remove this the code doesn't fail
with_resources(
[my_table_asset], resource_defs = resources_by_env[os.getenv("DAGSTER_DEPLOYMENT")]
),
]r
rex
12/20/2022, 11:02 PMAnd you’re running into this during local development? Do you have any stack trace to work with here?
This minimal example works for me, running
dagit -f ./minimal_example.py Copy code
# ./minimal_example.py
from dagster import (
AssetSelection,
DailyPartitionsDefinition,
ScheduleDefinition,
asset,
define_asset_job,
repository,
)
partitions_def = DailyPartitionsDefinition(start_date="2022-11-01")
@asset(
group_name="staging",
partitions_def=partitions_def,
)
def my_asset():
pass
# Materialize all assets in the repository
partitioned_asset_job = define_asset_job(
"partitioned_asset_job",
AssetSelection.assets(my_asset),
partitions_def=partitions_def,
)
@repository
def minimal_example():
return [
ScheduleDefinition(
job=partitioned_asset_job,
cron_schedule="@weekly",
),
my_asset,
]l
Lindsay S
12/20/2022, 11:10 PMhm... yeh, I must think there must just be an error somewhere... i'll look for it and grab the stack trace if I can't identify. Thanks for this.
I wonder if I was turned around somewhere because the docs seem to have shifted to the new Definitions api and I am still working with repositories, not sure why that would change much, though...
r
rex
12/20/2022, 11:12 PMThe introduction of the
Definitions@repositoryit was a backwards compatible change
l
Lindsay S
12/20/2022, 11:13 PMoh yeh, i meant more of a user error issue reading through the doc. sorry for that confusion
r
rex
12/20/2022, 11:15 PMHere’s the equivalent minimal example using
Definitions Copy code
from dagster import (
AssetSelection,
DailyPartitionsDefinition,
Definitions,
ScheduleDefinition,
asset,
define_asset_job,
)
partitions_def = DailyPartitionsDefinition(start_date="2022-11-01")
@asset(
group_name="staging",
partitions_def=partitions_def,
)
def my_asset():
pass
partitioned_asset_job = define_asset_job(
"partitioned_asset_job",
AssetSelection.assets(my_asset),
partitions_def=partitions_def,
)
defs = Definitions(
assets=[my_asset],
schedules=[
ScheduleDefinition(
job=partitioned_asset_job,
cron_schedule="@weekly",
)
],
)thank you box 1
m
marcos
01/29/2023, 10:05 PM@rex My code is very similar to what you put here with the exception that I am using static partitions. The job runs great when I trigger it manually. I can backfill partitions fine. However, when the schedule runs, it fails
dagster._check.CheckError: Failure condition: Tried to access partition_key_range for a non-partitioned run8 Views