Parallel execution on single machine?

[liezl200]

Can I run with parallel execution on a single machine? This is for an offline workflow, doesn't need to be deployed, just can be manually run in "local" mode from the devserver that dagster is installed on.

I've already set up a dagster pipeline but couldn't figure out how to configure parallel execution

Alternatives:
The alternative hacky workflow for parallelizing is to use a python script to generate and run 20 different tmux commands. But we'd miss out on retries, error message, nice UI that dagster provides.

Airflow I think supports a parallel executor config.

[catherinewu]

Hi @liezl200 👋 thanks for asking and sorry this isn't more discoverable in our docs! Yes, it's easy to run solids in parallel. All that's needed is to add this snippet to your run configuration (feel free to change the number 4):

execution:
  multiprocess:
    config:
      max_concurrent: 4

Please let me know if that's what you're looking for!

docs: https://docs.dagster.io/_apidocs/execution#dagster.multiprocess_executor

[sryza]

I believe there's one more step, which is to make sure that the pipeline is using an IO manager that writes out intermediate values to the filesystem instead of storing them in memory. Without that step, solids won't be able to access the outputs from upstream solids.

E.g. something like this:

from dagster import ModeDefinition, fs_io_manager

@pipeline(mode_defs=[ModeDefinition(resource_defs={"io_manager": fs_io_manager})])
def my_pipeline():
    ...

[catherinewu]

yes, thanks for catching!

[liezl200]

There's an input error for fs_io_manager. Is there another import statement I should be using @sryza ? I tried:

1. from dagster import fs_io_manager as per https://docs.dagster.io/_apidocs/io-managers#dagster.fs_io_manager
2. import dagster ... Use dagster.fs_io_manager in @pipeline decorator
3. from dagster.core.storage import fs_io_manager
4. from dagster.core.storage.fs_io_manager import fs_io_manager

I was on
[[package]]
name = "dagit"
version = "0.9.14"
description = "Web UI for dagster."
category = "main"
optional = false
python-versions = "*"

[package.dependencies]
click = ">=7.0"
dagster = "0.9.14"
dagster-graphql = "0.9.14"
flask = ">=0.12.4

Updated and had the same issue
[[package]]
name = "dagit"
version = "0.9.22.post0"
description = "Web UI for dagster."
category = "main"
optional = false
python-versions = "*"

[package.dependencies]
click = ">=7.0"
dagster = "0.9.22.post0"
dagster-graphql = "0.9.22.post0"

dagster.core.errors.DagsterUserCodeProcessError: (DagsterImportError) - dagster.core.errors.DagsterImportError: Encountered ImportError: cannot import name 'fs_io_manager' from 'dagster' (/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/init.py)while importing module preconception from file /Users/lie/preconception/preconception.py. Local modules were resolved using the working directory/Users/lie/preconception. If another working directory should be used, please explicitly specify the appropriate path using the -dor--working-directoryfor CLI based targets or theworking_directoryconfiguration option forpython_file`-based workspace.yaml targets.

Stack Trace:
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/grpc/server.py", line 343, in ListRepositories
self._repository_symbols_and_code_pointers.loadable_repository_symbols,
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/grpc/server.py", line 105, in loadable_repository_symbols
self.load()
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/grpc/server.py", line 95, in load
self._loadable_repository_symbols = load_loadable_repository_symbols(
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/grpc/server.py", line 119, in load_loadable_repository_symbols
loadable_targets = get_loadable_targets(
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/grpc/utils.py", line 25, in get_loadable_targets
else loadable_targets_from_python_file(python_file, working_directory)
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/cli/workspace/autodiscovery.py", line 11, in loadable_targets_from_python_file
loaded_module = load_python_file(python_file, working_directory)
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/core/code_pointer.py", line 114, in load_python_file
six.raise_from(
File "", line 3, in raise_from

The above exception was the direct cause of the following exception:
(ImportError) - ImportError: cannot import name 'fs_io_manager' from 'dagster' (/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/init.py)

Stack Trace:
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/core/code_pointer.py", line 95, in load_python_file
return import_module_from_path(module_name, python_file)
File "/Users/lie/Library/Caches/pypoetry/virtualenvs/preconception-pSFOPY-K-py3.8/lib/python3.8/site-packages/dagster/seven/init.py", line 118, in import_module_from_path
spec.loader.exec_module(module)
File "", line 783, in exec_module
File "", line 219, in _call_with_frames_removed
File "/Users/lie/preconception/preconception.py", line 5, in
from dagster import ModeDefinition, fs_io_manager
`

cc @catherinewu @sryza

[alangenfeld]

fs_io_manager was introduced in 0.10.0 so you would need to upgrade to be able to use that.

The 0.9.X approximate equivalent would be including

intermediate_storage:
  filesystem:

so in total

execution:
  multiprocess:
intermediate_storage:
  filesystem:

(assuming you want the default concurrency otherwise you can set the config: max_concurrent: section)