Giacomo Colombo
03/11/2024, 3:43 PMinstance = DagsterInstance.get()
job_sample = define_asset_job(name="job_sample", selection=["my_asset_1", "my_asset_2"])
result = execute_job(reconstructable(job_sample), instance=instance)
Scout
03/11/2024, 3:43 PMjob
decorator and use it with the execute_job
method, you should define your job within a function that returns a JobDefinition
. This is necessary because when you use reconstructable
, it expects a function or definition produced by a decorated function, not an UnresolvedAssetJobDefinition
. Here's a step-by-step guide on how to properly create an Asset Job using the job
decorator and execute it: 1. Define your assets. 2. Create a function that returns a JobDefinition
by calling to_job()
on a graph or using the job
decorator. 3. Use the reconstructable
function to make your job reconstructable. 4. Execute the job using the execute_job
method. Here's an example based on the documentation: python from dagster import job, reconstructable, execute_job, DagsterInstance # Define your assets here # ... @job def job_sample(): # Define the job logic here by invoking assets or ops ... # Make the job reconstructable reconstructable_job_sample = reconstructable(job_sample) # Get the Dagster instance instance = DagsterInstance.get() # Execute the job result = execute_job(reconstructable_job_sample, instance=instance)
In this example, job_sample
is a function decorated with @job
, which makes it a JobDefinition
. Then, reconstructable(job_sample)
is used to create a reconstructable job that can be passed to execute_job
. If you encounter the DagsterInvariantViolationError
stating that the reconstructable target should be a function or definition produced by a decorated function, it means that you are likely passing an UnresolvedAssetJobDefinition
directly to reconstructable
, which is not supported. Instead, you should ensure that you are passing a function that returns a JobDefinition
or a JobDefinition
produced by a decorated function, as shown in the example above.