How can I run asset_job from python script and command line just like UI?

[takerfume]

Hi, I would like to know hot to run asset_job from python script and command.

I implemented the following script in tutorials/__init__.py .

And I don't know how to images_job from python script and command line.
I wouldn't like to run dagster ui or web server and I just would like to run job.

I tried running it by call load_assets_from_modules and materialize from ./run_materialize.py but materialize did not execute in parallel when using dynamic output to create a parallelized DAG.

When I executed images_job from the UI, the parallel processing occurred as expected in the parts where the DAG is parallelized.

I would like to run parallel dag execution by python script or command line.

• tutorials/__init__.py

from dagster import (
    asset,
    define_asset_job,
    AssetSelection,
    Definitions,
    FilesystemIOManager,
    op,
    # graph_asset,
    graph,
    DynamicOut,
    DynamicOutput,
    AssetsDefinition,
    AssetKey,
)

from typing import List, Any
import random

import time


@asset()
def example_asset_0() -> int:
    return 1


@asset()
def example_asset_1(example_asset_0: int) -> int:
    return example_asset_0 + 1


@asset()
def example_asset_2(example_asset_1: int) -> int:
    return example_asset_1 + 1


@op(out=DynamicOut(int))
def return_dynamic(input_val: int):
    print(input_val)
    # outputs = []
    for idx, page_key in enumerate(range(random.randint(5, 10))):
        yield DynamicOutput(page_key, mapping_key=str(idx))


@op()
def op1(value: int) -> int:
    time.sleep(10)
    return value


@op()
def op2(value: int) -> int:
    return value


@op()
def path_through(collected_list: List[Any]) -> List[Any]:
    return collected_list


@graph()
def graph_asset_example(input_val: int):
    result = path_through(return_dynamic(input_val).map(op1).map(op2).collect())
    return result


dynamic_graph_asset = AssetsDefinition.from_graph(
    graph_asset_example,
    keys_by_input_name={"input_val": AssetKey("example_asset_2")},
    keys_by_output_name={"result": AssetKey("dynamic_graph_asset_yeah")},
    metadata_by_output_name={"result": {"num_records": 1}},
)


images_job = define_asset_job(name="images_job", selection=AssetSelection.all())

defs = Definitions(
    assets=[
        example_asset_0,
        example_asset_1,
        example_asset_2,
        dynamic_graph_asset,
    ],
    jobs=[
        images_job,
    ],
    resources={
        "fs_io_manager": FilesystemIOManager(),
    },
)

• ./run_materialize.py
  • this cannot run dag with multiprocess.

import tutorial

from dagster import load_assets_from_modules, materialize

all_assets = load_assets_from_modules([tutorial])
materialize(all_assets)

• I also tried some cli but failed
  • command: dagster job execute --package-name tutorial -a images_job
  • error message:

(dagster_tutorials) yoshidataketo@YoshidanoMBP tutorial % dagster job execute --package-name tutorial -a images_job
Traceback (most recent call last):
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_cli/__init__.py", line 48, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/click/core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_cli/job.py", line 329, in job_execute_command
    execute_execute_command(instance, kwargs)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/telemetry.py", line 168, in wrap
    result = f(*args, **kwargs)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_cli/job.py", line 342, in execute_execute_command
    job_origin = get_job_python_origin_from_kwargs(kwargs)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_cli/workspace/cli_target.py", line 539, in get_job_python_origin_from_kwargs
    repo_definition = recon_repo.get_definition()
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/definitions/reconstruct.py", line 119, in get_definition
    return repository_def_from_pointer(self.pointer, self.repository_load_data)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/definitions/reconstruct.py", line 739, in repository_def_from_pointer
    target = def_from_pointer(pointer)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/definitions/reconstruct.py", line 650, in def_from_pointer
    return _check_is_loadable(target)  # type: ignore
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/definitions/reconstruct.py", line 605, in _check_is_loadable
    raise DagsterInvariantViolationError(
dagster._core.errors.DagsterInvariantViolationError: Loadable attributes must be either a JobDefinition, GraphDefinition, or RepositoryDefinition. Got UnresolvedAssetJobDefinition(name='images_job', selection=<dagster._core.definitions.asset_selection.AllSelection object at 0x1068755a0>, config=None, description=None, tags={}, metadata={}, partitions_def=None, executor_def=None, hooks=None).

[sryza]

Hey @takerfume - try replacing -a with -j:

dagster job execute --package-name tutorial -j images_job

When you use -a you're telling Dagster to execute the variable images_job. However, this variable refers to a job that's unresolved, because it's not inside a Definitions object with the assets it targets. By using -j, you load the entire module including the Definitions object, and then execute "images_job" within that set of definitions, after it has been bound to the assets.

[takerfume]

@sryza

Thank you for the clarification, it helped a lot, and the command you suggested worked perfectly.

I have 3 additional questions.

1. what kind of job can be executed with the -a option
   Is my understanding correct that if I define a job using the @job decorator, I can execute it with the -a option?

2. the details of how definitions work and how jobs are resolved and unresolved
   I'm looking to understand more about the mechanisms of definitions, and the resolving/unresolving of jobs in Dagster. Is there any documentation available that delves into the details of how definitions work and how jobs are resolved and unresolved?

3. Experimentation and Monitoring with Varying Parameters
   Additionally, during the development phase, I'd like to run jobs with various parameters to see which ones cause asset materialization to fail, and what the metadata for those materializations looks like. Is there a way to view the results of job executions from the dagster job execute command or a Python script within the Dagster development UI? Being able to do this would significantly streamline my experimentation process.
   It seems that only the executions initiated through the UI are visible in the execution history within the UI, which makes it challenging to run and monitor experiments with different parameters in parallel. Any advice on how to accomplish this would be greatly appreciated.

[sryza]

Is my understanding correct that if I define a job using the @job decorator, I can execute it with the -a option?

That's correct

I'm looking to understand more about the mechanisms of definitions, and the resolving/unresolving of jobs in Dagster. Is there any documentation available that delves into the details of how definitions work and how jobs are resolved and unresolved?

Alas I'm not aware of content on this, although I believe it would be valuable to add.

Additionally, during the development phase, I'd like to run jobs with various parameters to see which ones cause asset materialization to fail, and what the metadata for those materializations looks like. Is there a way to view the results of job executions from the dagster job execute command or a Python script within the Dagster development UI? Being able to do this would significantly streamline my experimentation process.

Mind posting a separate discussion for this question?

[takerfume]

Hi, thank you for your reply and I posted here.
#17704

[takerfume]

@sryza
Hi, sandy.

I would like to do the same thing as dagster job execute --package-name tutorial -j images_job by python script.

This is because I would like to implement config file not as yaml file but as python class.

I tried the following codes but all gets errors.

How should I solve this problem?

• errors are following.

from dagster import execute_job, DagsterInstance
from tutorial import images_job


execute_job(images_job, instance=DagsterInstance.get())

ended up with errors like

(dagster_tutorials) yoshidataketo@YoshidanoMacBook-Pro tutorial % pipenv run python run_job.py
Courtesy Notice: Pipenv found itself running within a virtual environment, so it will automatically use that environment, instead of creating its own for any project. You can set PIPENV_IGNORE_VIRTUALENVS=1 to force pipenv to ignore that environment and create its own instead. You can set PIPENV_VERBOSITY=-1 to suppress this warning.
No dagster instance configuration file (dagster.yaml) found at /Users/yoshidataketo/GitHub/dagster_tutorials/tutorial. Defaulting to loading and storing all metadata with /Users/yoshidataketo/GitHub/dagster_tutorials/tutorial. If this is the desired behavior, create an empty dagster.yaml file in /Users/yoshidataketo/GitHub/dagster_tutorials/tutorial.
Traceback (most recent call last):
  File "/Users/yoshidataketo/GitHub/dagster_tutorials/tutorial/run_job.py", line 5, in <module>
    execute_job(images_job, instance=DagsterInstance.get())
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-tUvzrG-S/lib/python3.11/site-packages/dagster/_core/execution/api.py", line 444, in execute_job
    check.inst_param(job, "job", ReconstructableJob)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-tUvzrG-S/lib/python3.11/site-packages/dagster/_check/__init__.py", line 625, in inst_param
    raise _param_type_mismatch_exception(
dagster._check.ParameterCheckError: Param "job" is not a ReconstructableJob. Got UnresolvedAssetJobDefinition(name='images_job', selection=<dagster._core.definitions.asset_selection.AllSelection object at 0x105a12b90>, config=None, description=None, tags={}, metadata={}, partitions_def=None, executor_def=None, hooks=None) which is type <class 'dagster._core.definitions.unresolved_asset_job_definition.UnresolvedAssetJobDefinition'>.

and
2.

from dagster import execute_job, DagsterInstance, reconstructable
from tutorial import images_job


execute_job(reconstructable(images_job), instance=DagsterInstance.get())

ended up with errors like

(dagster_tutorials) yoshidataketo@YoshidanoMacBook-Pro tutorial % pipenv run python run_job.py
Courtesy Notice: Pipenv found itself running within a virtual environment, so it will automatically use that environment, instead of creating its own for any project. You can set PIPENV_IGNORE_VIRTUALENVS=1 to force pipenv to ignore that environment and create its own instead. You can set PIPENV_VERBOSITY=-1 to suppress this warning.
Traceback (most recent call last):
  File "/Users/yoshidataketo/GitHub/dagster_tutorials/tutorial/run_job.py", line 5, in <module>
    execute_job(reconstructable(images_job), instance=DagsterInstance.get())
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-tUvzrG-S/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 406, in reconstructable
    raise DagsterInvariantViolationError(
dagster._core.errors.DagsterInvariantViolationError: Reconstructable target should be a function or definition produced by a decorated function, got <class 'dagster._core.definitions.unresolved_asset_job_definition.UnresolvedAssetJobDefinition'>.

and
3.

from dagster import execute_job, DagsterInstance, reconstructable
from tutorial import defs


def return_job_def():
    return defs.get_job_def(name="images_job")


execute_job(reconstructable(return_job_def), instance=DagsterInstance.get())

ended up with errors like (in this case, I can run job, but runtime error occurred.)

(dagster_tutorials) yoshidataketo@YoshidanoMacBook-Pro tutorial % pipenv run python run_job.py            
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - 33da4c03-96db-4689-a138-78c5c7bf4e37 - 4553 - RUN_START - Started execution of run for "images_job".
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - 33da4c03-96db-4689-a138-78c5c7bf4e37 - 4553 - ENGINE_EVENT - Executing steps using multiprocess executor: parent process (pid: 4553)
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - 33da4c03-96db-4689-a138-78c5c7bf4e37 - 4553 - example_asset_0 - STEP_WORKER_STARTING - Launching subprocess for "example_asset_0".
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - b4eb4d00-055b-400f-9c66-2141e80eb572 - 4558 - RUN_START - Started execution of run for "images_job".
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - b4eb4d00-055b-400f-9c66-2141e80eb572 - 4558 - ENGINE_EVENT - Executing steps using multiprocess executor: parent process (pid: 4558)
2023-11-13 17:06:46 +0900 - dagster - DEBUG - images_job - b4eb4d00-055b-400f-9c66-2141e80eb572 - 4558 - example_asset_0 - STEP_WORKER_STARTING - Launching subprocess for "example_asset_0".
2023-11-13 17:06:46 +0900 - dagster - ERROR - images_job - b4eb4d00-055b-400f-9c66-2141e80eb572 - 4558 - RUN_FAILURE - Execution of run for "images_job" failed. An exception was thrown during execution.

RuntimeError: 
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

Stack Trace:
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/execution/api.py", line 766, in job_execution_iterator
    for event in job_context.executor.execute(job_context, execution_plan):
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/executor/multiprocess.py", line 249, in execute
    event_or_none = next(step_iter)
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/executor/multiprocess.py", line 360, in execute_step_out_of_process
    for ret in execute_child_process_command(multiproc_ctx, command):
  File "/Users/yoshidataketo/.local/share/virtualenvs/dagster_tutorials-0ZYUpUs2/lib/python3.10/site-packages/dagster/_core/executor/child_process_executor.py", line 157, in execute_child_process_command
    process.start()
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/process.py", line 121, in start
    self._popen = self._Popen(self)
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/context.py", line 288, in _Popen
    return Popen(process_obj)
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/popen_spawn_posix.py", line 32, in __init__
    super().__init__(process_obj)
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/popen_fork.py", line 19, in __init__
    self._launch(process_obj)
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/popen_spawn_posix.py", line 42, in _launch
    prep_data = spawn.get_preparation_data(process_obj._name)
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/spawn.py", line 154, in get_preparation_data
    _check_not_importing_main()
  File "/Users/yoshidataketo/.pyenv/versions/3.10.12/lib/python3.10/multiprocessing/spawn.py", line 134, in _check_not_importing_main
    raise RuntimeError('''

[takerfume]

I reposted the above additional question in this another discussion (#17943).

[sryza]

Reopened discussion because closed discussions don't appear in search results