Hey folks. Trying out the Memoizable IO Manager and running into what seems like potentially a bug in output creation. I've got this sample code (pasting in thread) using

versioned_filesystem_io_manager

, my Solid has a version property defined, but I'm consistently seeing:

dagster.check.ParameterCheckError: Param "context.version" is not a str. Got None which is type <class 'NoneType'>.

Hello! I'm working on a race-condition issue. High level: I have a pipeline that starts with 2 solids that can run in parallel when using multiprocess executor. They each depend on an mlflow resource, which, when initialised, will create a new run if one doesn't exist yet, else use the one that already exists. Unfortunately, this is not a single call and so it is possible that both solids decide an mlflow run doesn't exist yet, and each create one. I have a couple of solutions in mind and would be interested to hear your thoughts if there are others: 1. There are

DEBUG

logs available

2021-07-12 14:02:44 - dagster - DEBUG - multiprocess_pipeline - 845b164c-932e-4c7e-9f6b-85877d5b5f24 - 78555 - dog - ENGINE_EVENT - Starting initialization of resources [io_manager, mlflow].

- somehow subscribe to these logs and if a solid sees a log like this then wait for it to get the respective

2021-07-12 14:02:45 - dagster - DEBUG - multiprocess_pipeline - 845b164c-932e-4c7e-9f6b-85877d5b5f24 - 78555 - dog - ENGINE_EVENT - Finished initialization of resources [io_manager, mlflow].

2. Is there a better source of events (e.g. ENGINE_EVENT) that can be subscribed to maybe? To do the above, but with an event rather than looking at logs 3. Is there a way to restrict parallelisation of resource initialisation by any chance? Or set some kind of minimum time elapsed between executing 2 solids? 4. hacky - sleep a random fraction of a second. Hopefully in our case this will usually solve the problem 5. ugly - do a second check after creating the run to see if 2 have been created and if so, delete the one that was created later

Is there a high-level release timeline available? Specifically wondering if a v1.0 release at a

Q1|Q2|.. of year YYYY

level can be shared at this time.

Using the configured api, is it possible to configure a solid in two separate places? For example, using the dbt solids, could I configure the solids to use a shared config for the project/profiles dir like this:

run_dbt_models = dbt_cli_run.configured(default_dbt_config, name="run_dbt_models")

And then in each of my pipelines, "reconfigure" the solids to add additional configs like so:

run_daily_models = configured(run_dbt_models, name="run_daily_models")({"models": ["+tag:daily"]})

(I ask because I am trying and failing to do this currently)

In the docs on configuring a k8s deployment, it mentions configuring an external postgres DB here. Specifically this section: "If you use a secrets manager like Vault, it may be convenient to manage this Secret outside of the Dagster Helm chart. In this case, the generation of this Secret within the chart should be disabled, and

.Values.global.postgresqlSecretName

should be set to the name of the externally managed Secret." Is there any more info on using an external secret? We use AWS secrets manager for our secrets, so I'll like to leverage that if possible.

Hello, got a quick question. Are there any testing examples (preferably with invocation) of

async

solids? currently working through this and there is a

DagsterTypeCheckDidNotPass

with type failure along the lines of

expected return_type got coroutine

. Our tests that use

execute_solid

for our async solids still work fwiw

Hello, need some clarification with some implementation details while reading the celery guide (https://docs.dagster.io/deployment/guides/celery#4-dagster-code). It’s noted for dagster code that: “Finally, you will need to make sure that the Dagster code you want the workers to execute is present in their environment, and that it is in sync with the code present on the node running Dagit. The easiest way to do this is typically to package your code into a Python module, and for the 

workspace.yaml

 you use to run Dagit to load from that module.” Is this saying that the celery workers should load from the same python modules that are specified in the workspace.yaml that Dagit uses, or that there is a way to configure a workspace.yaml for a dagster celery worker?

Hi, how do I force Dagit to stop forcing to scaffold solid input config, when using a Custom IO manager and the assets already exist, but needing the solids in place purely for run dependencies and order of execution

Is it possible to trigger a RunRequest with a preset configuration from a sensor?

Has anyone been successful in getting a fan out solid to work with dagster_dask? My experience has been that while the fan out gets executed successfully, the solid indicated by the

.map()

function isn't called, and neither do any of the subsequent solids in the graph. The graph works properly when not running on dask executor

Hi dagsters! I was asking earlier about control of solids order execution when using composite_solid on DynamicOutput The suggestion was to use tag

dagster/priority

. So I tried and thats works only for

InProcessExecutor

and

MultiprocessExecutor

with

max_concurrent: 1

but for

max_concurrent > 1

that will not work as well as for

CeleryExecutor

with

dagster-celery/priority

tag So tag

dagster-celery/priority

in case of CeleryExecutor with only one process (

autoscale=1,1

) have no effect in that particular example case Can anyone know how to achieve same effect on CeleryExecutor with one worker to execute all parts of composite_solid on first DynamicOutput and only after - on second

Hey there, is it normal for the daemon to sit in CrashLoop in Kubernetes? Or does it indicate an issue? If so, how to fix? Thanks!!!!!!!!!!!!!!!!!!!!!!!!!

Hi all! I think I might have found a small bug - or I need some help - when applying a success hook to a pipeline I lose my pipeline's description! 😱 I think this is new/different behaviour with 0.12.1. Code to reproduce in thread.

Hi all, I posted about this a while back - ive been picking this up on and off. I am trying to use Dagster to execute jobs on Data Bricks.

@pipeline(mode_defs=[ModeDefinition(resource_defs={"databricks_client": databricks_client})])
def spark_job():
    data_bricks_job_solid = create_databricks_job_solid(name="daily_stock")
    data_bricks_job_solid()

This code works. How do I add config to databricks for the job? I tried:

data_bricks_job_solid(
        {
            "name": "SparkPi Python job",
            "new_cluster": {
                "spark_version": "7.3.x-scala2.12",
                "node_type_id": "i3.xlarge",
                "num_workers": 2
            },
            "spark_python_task": {
                "python_file": "dbfs:/docs/pi.py",
                "parameters": [
                    "10"
                ]
            }
        }
    )

But I get an exception

Great presentation! Super excited for the changes. was wondering if the presentation will be uploaded anywhere?

Curious, is there a particular reason why setting the

DAGSTER_HOME

environment variable isn't explicitly mentioned in the "Getting Started" or "Tutorial" docs? I understand that there is a warning, but it is at run or execution time.

Hi folks, I want to dynamically register new repos. Wondering if there a guide on doing that? The scheme I have in mind. New Dagit api server come online -> Self register on some service discovery DB (say, Dynamo, ZK, etc.) -> The main fleet of dagit UI polls for a list of new repo servers -> register new repo What you think?

Hi All, At AppliedAI, We are building a machine learning framework based on Dagster. We have a question what is the best practice for developing, working with inference pipelines in Dagster? We would like it to have low latency, even for single sample inference. We would like to be able to perform inference in parallel. Ideally we would share the same model across multiple inference workflows. The inference pipeline should re-use solids from a separate ML training pipeline - We have tried developing pipelines which should be triggered based on a sensor which checks for new inputs(currently we tried using a root solid with ‘DynamicOutput’) - Behavior of

map()

on solids with dynamic outputs seems to collect all results first, only then apply any downstream solids (preventing infinite event streams) is this the expected behaviour? We have a few concerns: The time to instantiate a workflow for a single sample infernce could be quite slow. (can we keep workflows alive for multiple runs?) The workflows are blocking so we need multiple workflows, (are celery worker the solution to this?) Do you have any suggestions, ideas how we should be thinking about this?

Ideally I want one folder of solids (to minimise code duplication), for use in different combinations across pipelines/repos with each repo having a different virtual environment. Is there a way to achieve this with workspace.yaml?

Hi, we uninstalled and reinstalled our Dagster Helm chart (due to multiple daemons running, but thats not relevant to this). But, it's broken our aws-load-balancer ingress gateway. Does the ingress/loadbalancer need to be installed second, after dagit to work? I.e., should we uninstall and reinstall this too? Or , what else could be the issue (url now hits '502 bad gateway' or 504 gateway timeout). Thanks!!

hi team, context: trying to set up the multi container docker deployment locally. my problem: env vars are not getting passed to the actual container the would run the solid my configs in the 🧵 thanks for the help

Hi everyone! I'm struggling to get the Celery executor to work locally (following along with https://docs.dagster.io/deployment/guides/celery#execute-on-celery). I can start the dagster-celery worker fine and see that it's registered in RabbitMQ. When I launch the workflow from dagit, I get an error due to Celery not being imported:

2021-07-15 17:46:47 - dagster - DEBUG - parallel_pipeline - f3b33113-c28d-47d5-a5dc-541b60912777 - 159937 - PIPELINE_START - Started execution of pipeline "parallel_pipeline".
2021-07-15 17:46:47 - dagster - ERROR - parallel_pipeline - f3b33113-c28d-47d5-a5dc-541b60912777 - 159937 - PIPELINE_FAILURE - Execution of pipeline "parallel_pipeline" failed. An exception was thrown during execution.

ModuleNotFoundError: No module named 'celery.exceptions'; 'celery' is not a package

Stack Trace:
  File "/home/adriano/playground/dagster/dagster-demo/.venv/lib/python3.9/site-packages/dagster/core/execution/api.py", line 756, in pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/home/adriano/playground/dagster/dagster-demo/.venv/lib/python3.9/site-packages/dagster_celery/executor.py", line 157, in execute
    from .core_execution_loop import core_celery_execution_loop
  File "/home/adriano/playground/dagster/dagster-demo/.venv/lib/python3.9/site-packages/dagster_celery/core_execution_loop.py", line 4, in <module>
    from celery.exceptions import TaskRevokedError

The

dagster-celery

and

celery

(I tried both v4.4.6 and the latest v5.1.2) packages are installed in the virtualenv (I can even set a breakpoint on the offending import statement and import the module in my debug console just fine; the

site-packages

of the venv is on

sys.path

). Do you have any ideaswhy this import might fail or how I might further debug the issue? Thanks a ton! D

Hello everyone! While loading the schedules I am hitting the following error: Any thoughts on how to approach/fix this?

I’m doing a fresh install of Dagster and Dagit (0.12.1) in a new venv and getting

There are incompatible versions in the resolved dependencies:
  graphql-core<3,>=2.0 (from graphql-ws==0.3.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))
  graphql-core<3,>=2.1 (from dagster-graphql==0.12.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))
  graphql-core<3,>=2.1 (from flask-graphql==2.0.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))
  graphql-core<3,>=2.3 (from graphql-server-core==1.2.0->flask-graphql==2.0.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))
  graphql-core<3,>=2.3.2 (from gql==2.0.0->dagster-graphql==0.12.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))
  graphql-core<4,>=3.1.2 (from graphene==3.0b7->dagster-graphql==0.12.1->dagit==0.12.1->-r /tmp/pipenvvg9eykvbrequirements/pipenv-vug0hh0m-constraints.txt (line 13))

this happened when Flask published a breaking release about two months ago, could this be the reason again?

hey, me again. context: trying to set up the multi container docker deployment locally. my problem:

This engine is only compatible with a CeleryK8sRunLauncher; configure the CeleryK8sRunLauncher on your instance to use it.

my mode def:

return ModeDefinition(
        name="mode",
        resource_defs=resources,
        executor_defs=[celery_k8s_job_executor] if prod else default_executors,
    )

what am I doing wrong ?

Trying to test a schedule following the example here, but getting:

dagster.core.errors.DagsterInvalidInvocationError: Schedule decorated function has context argument, but no context argument was provided when invoking.

Here’s the code. I know that

validate_run_config

is currently experimental. So maybe this is just a bug that needs some work.

@schedule(
    pipeline_name="collective_pipeline",
    cron_schedule="0 7 * * *",
    execution_timezone="UTC",
    mode='prod'
)
def coll_daily_schedule():
    return {}


def test_coll_daily_schedule():
    run_config = coll_daily_schedule()
    assert validate_run_config(collective_pipeline, run_config)

Can someone please help me understand this error?

Missing required config entry "solids" at the root. Sample config for missing entry: {'solids': {'fit_predict_inspect': {'inputs': {'use_dates': '<selector>'}}}}

Whats the best way to run

dbt deps

in dagster?

Hi, I am newbie to dagster, I got dagster daemon & dagit up & running, I was wondering how can I run user deployments in their own namespaces, and also user jobs/pipelines in their own namespaces ?

Can I import ColumnConstraintViolationException from anywhere? Thought it'd be in dagster_pandas? Below is an example from the docs. Looked like there'd be a nice in-built solution to then display the data that fails the check.