set the channel description: For discussion around using Dagster with Dask

👋 thought it would be useful to have a home for Dask-related discussion

@King Chung Huang thank you for the recent contributions to dagster-dask! given the recent activity from a few contributors w/ dagster + dask, wanted to get a channel up for any ad-hoc discussion

Cool! We’re starting to use Dask and Dagster together quite a bit here, and developed some code to enable it. Definitely interested in contributing more of our internal changes back to the Dagster project.

that’s fantastic

we’re still working on our process for getting PRs reviewed and merge, but will get someone to take a look within the next few days

I'm trying to execute a user code deployment pipeline using a dask cluster but I haven't been successful yet. I deployed a Dask cluster using their helm chart and then added the dns of the dask scheduler to my run config as described in the docs. When executing a pipeline I get the error:

ModuleNotFoundError: No module named 'dagster_dask'
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/api.py", line 380, in _execute_run_command_body
    for event in execute_run_iterator(recon_pipeline, pipeline_run, instance):
  File "/usr/local/lib/python3.8/site-packages/dagster/core/execution/api.py", line 727, in __iter__
    for event in self.iterator(
  File "/usr/local/lib/python3.8/site-packages/dagster/core/execution/api.py", line 665, in _pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/usr/local/lib/python3.8/site-packages/dagster_dask/executor.py", line 256, in execute
    for future, result in iterate_with_context(raise_interrupts_immediately, futures):
  File "/usr/local/lib/python3.8/site-packages/dagster/utils/__init__.py", line 443, in iterate_with_context
    next_output = next(iterator)
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 4475, in __next__
    return self._get_and_raise()
  File "/usr/local/lib/python3.8/site-packages/distributed/client.py", line 4466, in _get_and_raise
    raise exc.with_traceback(tb)
  File "/opt/conda/lib/python3.8/site-packages/distributed/protocol/pickle.py", line 75, in loads

I think this means I need to add the package 'dagster_dask' to my dask workers, is that correct?

Are there any detailed examples of running a map through dagster-dask? We want to use this as part of our pipeline but I can't see any sepcifics other than configuration. I'm wondering mostly how I would call the dagster solid with something to enumerate, or does it try to figure out what input object to enumerate?

Hey 👋 I’m wondering how folks are passing Dask dataframes between Dagster solids, specifically in the context of a multiprocess execution pipeline where I/O can’t be in-memory? Apart from writing/reading the computed DataFrame to disk between each solid, I tried writing an IO Manager that pickles the

<http://dd.to|dd.to>_delayed()

representation and loads it back as

dd.from_delayed()

but those still seemed to include

lambda

functions which Python’s

pickle

can’t handle. For context, the Dagster multiprocess bit is unrelated to Dask per-say; I’m enriching a subset of rows by querying an API which I’m attempting to do in batches to: 1. speed it up by processing multiple batches at a time 2. reduce the risk of losing everything in case of failure since API calls are paid — if 1 batch fails, I can simply replay that single batch.