[dask] PBSCluster - nothing is executed - pipeline freezes

[erwanbodere]

Hi,

I want to execute a pipeline with dagster on an HPC cluster using PBS Pro.

The configuration for PBS can be found here : https://github.com/dask/dask-jobqueue/blob/main/dask_jobqueue/jobqueue.yaml

I can execute a process with Dask on our HPC cluster but i don't achieve to execute a pipeline with dasgter. Dasgter freezes at "PIPELINE START - Started execution of pipeline", without any errors.

Here is my yaml file :

execution:
  dask:
    config:
      cluster:
        pbs:
          extra: [ "prejob1", "prejob2"]

Is ti possible to have more logs about Dask, for instance with cluster.job_script() in order to understand what's happen ?

See : https://jobqueue.dask.org/en/latest/debug.html

Regards

[erwanbodere]

Hi,
I have looked at the source code. The execution freezes at :

• dagster_dask.executor.py#276

            # Allow interrupts while waiting for the results from Dask
            for future, result in iterate_with_context(raise_execution_interrupts, futures):
                for step_event in result:
                    check.inst(step_event, DagsterEvent)
                    yield step_event

• dagster.utils.py#366

# Executes the next() function within an instance of the supplied context manager class
# (leaving the context before yielding each result)
def iterate_with_context(context, iterator):
    while True:
        # Allow interrupts during user code so that we can terminate slow/hanging steps
        with context():
            try:
                next_output = next(iterator)
            except StopIteration:
                return

        yield next_output

Nothing is executed at next(iterator).

Here are the steps to reproduce :

• dagster-pbs.py

from dagster import ModeDefinition, default_executors, fs_io_manager, pipeline
from dagster_dask import dask_executor
from dagster_shell import create_shell_command_solid


@pipeline(
    mode_defs=[
        ModeDefinition(
            resource_defs={"io_manager": fs_io_manager},
            executor_defs=default_executors + [dask_executor],
        )
    ]
)
def hello_pipeline():
    cmd_solid = create_shell_command_solid('echo "hello, world!"', name="hello_world")
    cmd_solid()

• dagster-pbs.yaml

execution:
  dask:
    config:
      cluster:
        pbs:
          memory: '24GB'
          cores: 1

loggers:
  console:
    config:
      log_level: DEBUG

[erwanbodere]

Hi,

Maybe dagster is not working with dask-jobqueue ?

The LocalCluster is using distributed.LocalCluster and not dask_jobqueue.local.LocalCluster.

Any idea ?

Rregards

[alangenfeld]

What information can you gather from the PBS cluster about what information it is receiving?

[mgasner]

When you say "The LocalCluster is using distributed.LocalCluster and not dask_jobqueue.local.LocalCluster.", how are you observing that? Also, which config are you using for the executor -- there are two snippets above

[erwanbodere]

In DaskExecutor.execute() (in dagster_dask.executory.py) :

• LocalCluster and SSHCluster are imported from dask.distributed (thus from distributed package)
• PBSCluster, SLURMCluster, ... are imported from dask_jobqueue
• others from specific packages (dask_yarn, dask_kubernetes)

If i try with PBSCluster on my computer (which doesn't have the qsub binary), the task is not executed. I should have had an error : qsub : command not found

The execution freezes at distributed.utils.py

async def All(args, quiet_exceptions=()):
    """Wait on many tasks at the same time

    Err once any of the tasks err.

    See https://github.com/tornadoweb/tornado/issues/1546

    Parameters
    ----------
    args: futures to wait for
    quiet_exceptions: tuple, Exception
        Exception types to avoid logging if they fail
    """
    tasks = gen.WaitIterator(*map(asyncio.ensure_future, args))
    results = [None for _ in args]
    while not tasks.done():
        try:
            result = await tasks.next() # Nothing is executed here
        except Exception:
          ...

Could you try with dask-jobqueue on you side ?

from dagster import ModeDefinition, default_executors, fs_io_manager, pipeline, solid
from dagster_dask import dask_executor


@solid
def hello_logs(context):
    context.log.info("Hello, world!")


@pipeline(
    mode_defs=[
        ModeDefinition(
            resource_defs={"io_manager": fs_io_manager},
            executor_defs=default_executors + [dask_executor],
        )
    ]
)
def hello_pipeline():
    hello_logs()

execution:
  dask:
    config:
      cluster:
        pbs:
          queue: sequentiel
          memory: '24GB'
          cores: 1