How to run asset job via python script.

[takerfume]

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 would not like to use materialize() or materialize_in_memory() because like to run DAG in multiprocess.
dagster job execute works well but inputting config file via yaml is not convenient for me.
My context is this discussion (#17675)

I tried the following codes but all gets errors.
How should I solve this problem?

implementation of 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(),
    },
)

errors I encounterred

This implementation of ./run_job.py

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
This implementation of ./run_job.py

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
This implementation of ./run_job.py

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('''

[alangenfeld]

What happens when you update your final entry to

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


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


if __name__ == '__main__':
    execute_job(reconstructable(return_job_def), instance=DagsterInstance.get())

I believe the error you are getting is from execute_job recursing when the file is imported since its not gated by that name == main check

[takerfume]

@alangenfeld

Thank you for your help! I tried your code and it worked perfectly.

This experience has reminded me again that Python's multiprocessing module, when creating a new process, reloads the script where the function to be executed in parallel is defined.(This mechanism is necessary to share code across processes in a multi-process execution. Document is here )

I realized that my initial code:

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

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

is equivalent to the following pattern, where both the function for multi-process execution and the call to execute it are in the same script. This was causing recursive behavior during multi-process execution, as it essentially calls itself.

def f(x):
    return x * x

with Pool(5) as p:
    print(p.map(f, [1, 2, 3]))

It was a great learning experience! Thank you once again.