["dagster","api","execute_run","{\"__class__\": \"ExecuteRunArgs\", \"instance_ref\": {\"__class__\": \"InstanceRef\", \"compute_logs_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"LocalComputeLogManager\", \"config_yaml\": \"base_dir: /opt/dagster/dagster_home/storage\\n\", \"module_name\": \"dagster.core.storage.local_compute_log_manager\"}, \"custom_instance_class_data\": null, \"event_storage_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"PostgresEventLogStorage\", \"config_yaml\": \"postgres_db:\\n db_name:\\n env: DAGSTER_POSTGRES_DB\\n hostname:\\n env: DAGSTER_POSTGRES_HOSTNAME\\n password:\\n env: DAGSTER_POSTGRES_PASSWORD\\n port: 5432\\n username:\\n env: DAGSTER_POSTGRES_USER\\n\", \"module_name\": \"dagster_postgres.event_log\"}, \"local_artifact_storage_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"LocalArtifactStorage\", \"config_yaml\": \"base_dir: /opt/dagster/dagster_home/\\n\", \"module_name\": \"dagster.core.storage.root\"}, \"run_coordinator_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"QueuedRunCoordinator\", \"config_yaml\": \"{}\\n\", \"module_name\": \"dagster.core.run_coordinator\"}, \"run_launcher_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"EcsRunLauncher\", \"config_yaml\": \"{}\\n\", \"module_name\": \"dagster_aws.ecs\"}, \"run_storage_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"PostgresRunStorage\", \"config_yaml\": \"postgres_db:\\n db_name:\\n env: DAGSTER_POSTGRES_DB\\n hostname:\\n env: DAGSTER_POSTGRES_HOSTNAME\\n password:\\n env: DAGSTER_POSTGRES_PASSWORD\\n port: 5432\\n username:\\n env: DAGSTER_POSTGRES_USER\\n\", \"module_name\": \"dagster_postgres.run_storage\"}, \"schedule_storage_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"PostgresScheduleStorage\", \"config_yaml\": \"postgres_db:\\n db_name:\\n env: DAGSTER_POSTGRES_DB\\n hostname:\\n env: DAGSTER_POSTGRES_HOSTNAME\\n password:\\n env: DAGSTER_POSTGRES_PASSWORD\\n port: 5432\\n username:\\n env: DAGSTER_POSTGRES_USER\\n\", \"module_name\": \"dagster_postgres.schedule_storage\"}, \"scheduler_data\": {\"__class__\": \"ConfigurableClassData\", \"class_name\": \"DagsterDaemonScheduler\", \"config_yaml\": \"{}\\n\", \"module_name\": \"dagster.core.scheduler\"}, \"settings\": {\"telemetry\": null}}, \"pipeline_origin\": {\"__class__\": \"PipelinePythonOrigin\", \"pipeline_name\": \"[deleted]\", \"repository_origin\": {\"__class__\": \"RepositoryPythonOrigin\", \"code_pointer\": {\"__class__\": \"FileCodePointer\", \"fn_name\": \"my_repository\", \"python_file\": \"repo.py\", \"working_directory\": \"[deleted]\"}, \"container_image\": \[deleted]\", \"executable_path\": \"/opt/pysetup/.venv/bin/python\"}}, \"pipeline_run_id\": \"018515a6-02f2-4441-a23e-25cedb785ee0\"}"]

I deployed dagit using the deploy_ecs example. I can execute the example pipeline successfully, but not my own pipeline. I am using version 0.11.16. Does anybody has an idea what it wrong and how I could solve this issue?

Hi All, I am running into problems when launching a pipeline from dagit that is deployed on ECS. Dagit & Daemon seem to start a task based on the

dagster-run

task definition containing a container

run

, but it is killed due to the next error:

Error: Got unexpected extra arguments ("ExecuteRunArgs", "instance_ref": {"__class__": "InstanceRef", "compute_logs_data": {"__class__": "ConfigurableClassData", "class_name": "LocalComputeLogManager", "config_yaml": "base_dir: /opt/dagster/dagster_home/storage\n", "module_name": "dagster.core.storage.local_compute_log_manager"}, "custom_instance_class_data": null, "event_storage_data": {"__class__": "ConfigurableClassData", "class_name": "PostgresEventLogStorage", "config_yaml": "postgres_db:\n db_name:\n env: DAGSTER_POSTGRES_DB\n hostname:\n env: DAGSTER_POSTGRES_HOSTNAME\n password:\n env: DAGSTER_POSTGRES_PASSWORD\n port: 5432\n username:\n env: DAGSTER_POSTGRES_USER\n", "module_name": "dagster_postgres.event_log"}, "local_artifact_storage_data": {"__class__": "ConfigurableClassData", "class_name": "LocalArtifactStorage", "config_yaml": "base_dir: /opt/dagster/dagster_home/\n", "module_name": "dagster.core.storage.root"}, "run_coordinator_data": {"__class__": "ConfigurableClassData", "class_name": "QueuedRunCoordinator", "config_yaml": "{}\n", "module_name": "dagster.core.run_coordinator"}, "run_launcher_data": {"__class__": "ConfigurableClassData", "class_name": "EcsRunLauncher", "config_yaml": "{}\n", "module_name": "dagster_aws.ecs"}, "run_storage_data": {"__class__": "ConfigurableClassData", "class_name": "PostgresRunStorage", "config_yaml": "postgres_db:\n db_name:\n env: DAGSTER_POSTGRES_DB\n hostname:\n env: DAGSTER_POSTGRES_HOSTNAME\n password:\n env: DAGSTER_POSTGRES_PASSWORD\n port: 5432\n username:\n env: DAGSTER_POSTGRES_USER\n", "module_name": "dagster_postgres.run_storage"}, "schedule_storage_data": {"__class__": "ConfigurableClassData", "class_name": "PostgresScheduleStorage", "config_yaml": "postgres_db:\n db_name:\n env: DAGSTER_POSTGRES_DB\n hostname:\n env: DAGSTER_POSTGRES_HOSTNAME\n password:\n env: DAGSTER_POSTGRES_PASSWORD\n port: 5432\n username:\n env: DAGSTER_POSTGRES_USER\n", "module_name": "dagster_postgres.schedule_storage"}, "scheduler_data": {"__class__": "ConfigurableClassData", "class_name": "DagsterDaemonScheduler", "config_yaml": "{}\n", "module_name": "dagster.core.scheduler"}, "settings": {"telemetry": null}}, "pipeline_origin": {"__class__": "PipelinePythonOrigin", "pipeline_name": "[removed]", "repository_origin": {"__class__": "RepositoryPythonOrigin", "code_pointer": {"__class__": "FileCodePointer", "fn_name": "my_repository", "python_file": "repo.py", "working_directory": "[removed]"}, "container_image": "[removed]", "executable_path": "/opt/pysetup/.venv/bin/python"}}, "pipeline_run_id": "018515a6-02f2-4441-a23e-25cedb785ee0"})

ECS injects the next command in this

run

container:

Hello! Is there any way of accessing resources from a sensor? I would like to run a pipeline when a new elastic search document is uploaded to an index but I cannot access to the resource of ES I coded. Is there a way to avoid hard-coding the name of the index?

Hello, I'm having some issues when my pipeline runs on our K8 cluster but it works fine when running locally. Logs show this import error:

ImportError: cannot import name 'EventRecord' from 'dagster.core.events.log' (/root/.local/lib/python3.9/site-packages/dagster/core/events/log.py)

We are installing the following dagster files from the Docker file:

RUN pip install --user \
    dagster==${DAGSTER_VERSION} \
    dagster-graphql==${DAGSTER_VERSION} \
    dagster-postgres==${DAGSTER_VERSION} \
    dagster-cron==${DAGSTER_VERSION} \
    dagster-celery[flower,redis,kubernetes]==${DAGSTER_VERSION} \
    dagster-aws==${DAGSTER_VERSION} \
    dagster-k8s==${DAGSTER_VERSION} \
    dagster-celery-k8s==${DAGSTER_VERSION} \
    dagit==${DAGSTER_VERSION}

With

ARG DAGSTER_VERSION=0.11.3

Is there a package I'm missing?

Hello, I am working with an event-driven system built around kafka. I was wondering if there is a way to use Dagster to take advantage of the monitoring and logic provided by Dagster while keeping the usage as simple as possible.

Is there any other documentation somewhere with more examples using the solid hooks for slack? Currently reading https://docs.dagster.io/concepts/solids-pipelines/solid-hooks However, I'm missing the big picture on how this works with the

@daily_schedule

decorator. I'm unsure where to place the configs for slack for the token and channel information. Currently my pipeline is being called by

@daily_schedule(
    pipeline_name="my_pipeline",
    start_date=datetime(2021, 4, 1),
    execution_time=time(hour=8, minute=15),
    execution_timezone="UTC",
    mode="prod"
)
def my_schedule(date):
    return {
        "solids": {
            "fetch_data": {
                "config": {
                    "date": date.strftime("%Y-%m-%d")
                }
            },
            "delete_daily_data": {
                "config": {
                    "date": date.strftime("%Y-%m-%d")
                }
            },
            "validate_data": {
                "config": {
                    "date": date.strftime("%Y-%m-%d")
                }
            }
        }
    }

would I add another section within the return for resources?

Hi, I am running a Kubernetes deployment of Dagster, but I cannot see raw compute logs (stdout / stderr). I saw in helm’s

values.yaml

file that this stuff is handled by

computeLogManager

. Should I switch from

LocalComputeLogManager

to another one? perhaps

S3ComputeLogManager

? Or can I do something to make it work with ``LocalComputeLogManager`` itself?

With dynamic output orchestration coming out of experimental mode in 0.12.0, will it be possible to access

mapping_keys

in the context of downstream solids? (RE: https://dagster.slack.com/archives/C01U954MEER/p1620400766465200?thread_ts=1620397879.463400&cid=C01U954MEER)

Hey there, does anyone know how to grant systemserviceaccountdefault:dagster-release access to Opaque secrets in the kubernetes Helm deployment?

Hi team, I just wanted to get some thoughts on how to manage / organize the Assets and best practices on Asset Key naming. In our setup, we do have a set of aggregation pipelines which does aggregation on certain events and

group by

based on entity ids. Each pipeline also yields an asset for the respective aggregate. We also have downstream pipelines which depend on these aggregates and gets triggered based on the aggregates assets. Currently I am generating the asset keys by concatenating different strings like event name, aggregate name, entity ids, etc., For example:

n_customers_support_account_troubleshooting_tasks_SUM_user_id

. I feel that it can become so hard to manage them with such keys as the number of assets grow larger. Just wondering if the team has any thoughts or best practices towards asset management. Any plans to support tags for assets (similar to pipelines)?

Hi, I'm using the experimental 'metadata' param on

OutputDefinition

to provide hints to the custom IOManager we use. Not sure if this is the right channel, but I'd like to upvote it in the hope it stays. Thanks Paul

I’m trying to drive adoption of dagster in my team. I’ve had papermill running but since experimenting in a new environment I’ve had no luck. Keeps telling me dagstermill is not installed. But I’m almost certain it is.

Hi, I'd like to have both per-output IO Managers and multiple mode definitions, that will change those settings. The only solution I've found is to create IO Managers as aggregation of the possible ones and a config to switch to the right one at run time. Is there anything smarter in Dagster for that?

Hi, im playing with DynamicOutput and composite_solid, invoking composite_solid on DynamicOutput my pipeline looks like

dyn_outputs = produce_dyn_outputs()
dyn_outputs.map(compose_plus_two_mul_ten).collect()

is this equal to?

dyn_outputs = produce_dyn_outputs()
dyn_outputs.map(plus_two).map(mul_ten).collect()

where

compose_plus_two_mul_ten

is composite_solid consist of

plus_two

and

mul_ten

solids The question here is: How i can control the order of execution of that 2 solids? Why

plus_two

executes first on all outputs and only than execution of

mul_ten

started Use case: first solid will train model and second commit it to general dvc repo

Hi - I am running dagster in K8s using the K8sRunLauncher. Some of my solids require connecting to an MSSQL database using Kerberos authentication. To solve this, I need to customize the dagster k8s job to create a sidecar container (that will request and refresh the krb ticket). I have tried using the UserDefinedDagsterK8sConfig, but looking at the code, I don't think I can achieve what I want. I suppose I need to implement a CustomRunLauncher - Do you have any doc or example on how to do that?

Hello! I'm on dagster version 0.11.16. I am getting weird or unexpected behavior when I first launch my dagit instance and then choose a pipeline and then when I click on the "Playground" tab, the Chrome browser will show a completely blank white screen. I have to go back to dagit's home URL: http://127.0.0.1:3000 then repeat the steps, click on "Playground" tab, then all is normal. If I repeat again, the problem doesn't happen again. I will see if the problem will repeat if I clear the browser cache.

hello everyone, anyone successfully try this? especially

make_email_on_pipeline_failure_sensor

https://dagster.slack.com/archives/CG4QMQW0H/p1625783653190300

Hello everyone! I'm having a very weird issue. When deploying dagster locally on minikube it works perfectly but when I try to deploy it with the same configuration in my company's k8 cluster it goes slowly and when I try to run pipelines it appears "starting" forever. I get this message in the port-forward command:

E0709 11:02:52.896413  378418 portforward.go:372] error copying from remote stream to local connection: readfrom tcp6 [::1]:8080->[::1]:35554: write tcp6 [::1]:8080->[::1]:35554: write: broken pipe

I attach an image of the forever "starting" and the quota of the namespace (in case it's useful). Do you have any idea where the issue could be?

Hi everyone, I'm Vivek from Hyderabad, India. I recently deployed dagster using helm chart on Redhat openshift container platform 4.7 and ran into an error during the (example_usercode - dagster/*user-code-example:0.11.16*) execution. I actually ran into a permission denied issue (/opt/dagster/dagster_home/storage - mkdir) with Run_Coordinator (Job - Pods). Can anyone kindly help with this issue. Run Configuration: solids: multiply_the_word: config: factor: 2 inputs: word: value: test

Hello, I'm starting with Dagster, and have just tried with the very recently released (0.12 !) Dynamic Mapping & Collect. It works on my case, but the dynamic steps are executed one after another (like in only 1 thread). I've checked with /proc/cpuinfo, and the 4 cores are reported. Is there any setting or what could I have done wrong?

Hello I got a strange message on scheduled pipeline :

Hello I got a strange message on scheduled pipeline : FileNotFoundError: [Errno 2] No such file or directory: 'tail' and I not using any directory named 'tail' and tail command is present on system, but the stack trace not involve any of my pipeline or solid source code, any help please ?

We are trying to use a yaml file to specify run configuration for a pipeline run in a schedule (by loading it with

load_yaml_from_path

) but it doesn't seem to be parsed in the same way as a regular preset; the

env: ...

keys I use are rejected as invalid scalars. Any suggestions?

I'm working on deploying dagster to k8s using helm and I'm wondering how I can configure run storage, schedule storage, run coordinators, etc. In the docs, here, I see this section where the config can be passed as params. Is there a way to deploy the

dagster.yaml

file to configure these settings instead? Would it be built into the user code deployment image? (sorry if this is documented, I'm having trouble finding it).

Hello! Sorry in advance for the wall of text, tldr is in bold at the end. I’m looking for a system to orchestrate my data analysis workflows, and dagster seems like the best option except for one detail I’m struggling with. The most natural way for me to decompose my analysis into reusable functional steps/solids requires some simple logic at the dag/pipeline level to determine which steps need to be included and how they need to be connected. This logic would be simple loops and conditionals driven by configuration parameters that would be known at launch time, and it would not depend on any outputs or anything else that could change during runtime. When I look at the dagster documentation, I see that there are dynamic pipeline features for mapping solids over a dynamic output, but this only covers some of the variability that I’m looking for. This would also seem to require that I create extra solids purely for the purpose of converting configuration parameters into outputs, which gives me the impression that this feature is intended for things that vary during runtime instead of launch time. I also see features like lazy loaded pipeline definitions or reconstructable pipelines that defer pipeline generation until launch time, but those functions still don’t seem to have access to the context or configuration objects. There is an example in the docs showing a pipeline being programmatically built up from a yaml file, perhaps the most extreme version of dynamic pipeline definition possible, but even in that case the file being referencing is static/hardcoded.  I got excited when I saw that composite solids can take configuration, but I can’t figure out how to actually access that configuration when deciding which solids to run. The basic functionality I’m looking for here doesn’t seem like a strange ask, and it's so close to being provided by some of these features that I think I’m probably missing something obvious. Can anyone suggest a way to generate a pipeline as a function of the run configuration? Thanks!

Hi. I'm having an issue with the dagit's playground. When I start the server at my machine, to run my pipelines, the playground doesn't update, It's kind of frozen, I need to refresh the page manually to see what's the actual step of the pipeline that's currently running. Tried to set up everything from scratch in a new environment using pip but the problem persists (working from a mac). The funny thing is that when I set up dagster + dagit in a windows machine the dagit playground works perfectly. Did anyone had the same issue? Thanks for your attention!

Hi, is there a way to .map a @composite_solid to a dynamic output? All works if I use just @solid, but not @composite_solid. Error:

dagster.core.errors.DagsterInvalidDefinitionError: @composite_solid clean returned problematic value of type <class 'list'>. Expected return value from invoked solid or dict mapping output name to return values from invoked solids

Thanks for any advice!

guys, after deploying Dagit — is that normal to have those many

dagster.grpc

threads running?

Guys, I’ve successfully deployed multiple Dagit instances — I have a few questions regarding daemon for queuing runs: 1. Based on the doc, it seems like I should only spin up ONE and only daemon process, and ALL Dagit instances will know it’s health via reading heartbeat records from DB. Is this correct? 2. how does this one and only run queuing daemon work with those many Dagit instances? My guess is that whenever Dagit (any instance) got a request for triggering a run, it will first insert the run request into DB, and check whether there is a daemon running — if there is no daemon, then Dagit initialize the run immediately; if there is a daemon, Dagit will do nothing and daemon will read run requests from DB and doing queuing stuff (by controlling each run request’s status).

Hi all, I'm currently using Dagster 0.11.14 (had to pin this version because I was getting errors about unconnected inputs when upgrading to v0.11.15) and I am getting this error when passing a list of columns to a solid as part of the solid config:

TypeError: Argument 'obj' has incorrect type (expected list, got frozenlist)

This is what my config looks like:

solid_uw_get_target_metrics_columns = configured(
    solid_get_target_df_columns, name="solid_uw_get_target_metrics_columns"
)({"columns": ["field", "source", "value", "reliability"]})

And this is what the solid looks like:

@solid(
    config_schema={
        "columns": Field([str], is_required=True),
    }
)
def solid_get_target_df_columns(context, dfs: Dict[str, DagsterDataFrame]):
    """Solid that returns a Dictionary where every DataFrame has the columns specified from the original DataFrame

    Parameters:
    ---------
    dfs : Dict[str, DagsterDataFrame]
        Dictionary containing DagsterDataFrames to retrieve columns from
    columns : List[str]
        List of target columns

    Returns:
    -------
        dfs: Dict[str, DagsterDataFrame]
            Dictionary containing DagsterDataFrames with target columns
    """
    processed_dfs = {}
    for key, df in dfs.items():
        processed_dfs[key] = df[context.solid_config["columns"]]

    return processed_dfs

If I cast to

list(context.solid_config["columns"])

it works, otherwise the error will keep appearing. Any ideas?