hello, i'm using dagster with k8s run coordinator, for 3 pipelines that are python ops calling C++ code. I'm using a service account for the dagster daemon pod but the default one for the job pods (they don't need special permissions). However, now i need to create another pipeline that launches an apache spark (scala) job, also in kubernetes. the pipeline would be first creating a job description file (i would do this in python) then launching a spark job. the pod launching this job also needs a service account, so that it can launch the spark cluster for the job. Can i make it so that only this pipeline launches with a special service account and not the other existing pipelines ?

hey, I just tried to redeploy dagster in our gcloud cluster and am getting this error. It says in

dagit

that I should report it here

Operation name: InstanceSensorsQuery

Message: Attempted to deserialize class "InstigatorState" which is not in the whitelist. This error can occur due to version skew, verify processes are running expected versions.
Descent path: <root:dict>

Path: ["repositoriesOrError","nodes",0,"sensors"]

Locations: [{"line":17,"column":9}]

Stack Trace:
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executor.py", line 452, in resolve_or_error
    return executor.execute(resolve_fn, source, info, **args)
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executors/sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/schema/external.py", line 198, in resolve_sensors
    [GrapheneSensor(graphene_info, sensor) for sensor in sensors],
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/schema/external.py", line 198, in <listcomp>
    [GrapheneSensor(graphene_info, sensor) for sensor in sensors],
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/schema/sensors.py", line 55, in __init__
    self._external_sensor.get_external_origin_id()
  File "/usr/local/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 1689, in get_job_state
    return self._schedule_storage.get_job_state(job_origin_id)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/schedules/sql_schedule_storage.py", line 66, in get_job_state
    return self._deserialize_rows(rows[:1])[0] if len(rows) else None
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/schedules/sql_schedule_storage.py", line 37, in _deserialize_rows
    return list(map(lambda r: deserialize_json_to_dagster_namedtuple(r[0]), rows))
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/schedules/sql_schedule_storage.py", line 37, in <lambda>
    return list(map(lambda r: deserialize_json_to_dagster_namedtuple(r[0]), rows))
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 371, in deserialize_json_to_dagster_namedtuple
    check.str_param(json_str, "json_str"), whitelist_map=_WHITELIST_MAP
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 399, in _deserialize_json
    return unpack_inner_value(value, whitelist_map=whitelist_map, descent_path=_root(value))
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 430, in unpack_inner_value
    f'Attempted to deserialize class "{klass_name}" which is not in the whitelist. '

Are there any plans to more natively interface with a catalogue like: https://datahubproject.io?

Noticed that iris_classify.png under the "Expressing dependencies" section is not loading in the following page: https://docs.dagster.io/integrations/dagstermill#notebooks-as-ops Tried with Chrome and Edge

Question about temporary data: I’m in the middle of converting from home-grown workflow system to Dagster. Old system uses a temporary directory for run data that’s named after the workflow run id. I was able to replicate that using the

root_run_id

in Dagster. That worked fine locally, but I’m now testing our deployment in ECS and realized that if a job fails its container will disappear, temporary data with it. So, is there any way to persist temporary data (that can be picked up when a job is resumed) aside from returning it from the op?

Hey, I am trying to increase the observability of our dagster pipeline, and as much as I love the UI, I would like to integrate it with the rest of our alerting system. Does Dagster have any support for distributed tracing? I am generally thinking OpenCensus or OpenTelemetry

Hi! Is there a way to "type hint" dagster resources somehow?

Hi! I’m super excited about the asset + dbt support and I have been trying to configure some jobs but I am facing a few roadblocks. In a simplified way, my dagster code is the following:

assets = load_assets_from_dbt_manifest(
    json.load(open(os.path.join(DBT_PROJECT_DIR, "target", "manifest.json")))
)
build_dbt_models = build_assets_job("build_dbt_models", assets=assets, resource_defs=DEV_RESOURCES)

When I try to load the job though, I get errors whenever my dbt project contains seeds and snapshots. I have then modified the file

dagster_dbt/asset_defs.py

to replace

if node_info["resource_type"] == "model"

by

if node_info["resource_type"] in ["model","snapshot","seed"]

. I can now load my dagster job properly but then I have some errors and unexpected behaviors when trying to materialize my dbt project: 1. I get errors that my seeds are not materialized (

Core compute for op "dbt_project" did not return an output for non-optional output "raw_customers"

), most likely due to the fact that only a

dbt run

is done, and not a

dbt seed

2. If I go to the assets view and want to materialize only one of them, dagster still do a

dbt run --select asset1 asset2 assset3 ...

, not filtering down the model I asked to materialize 3. If I recall correctly, in previous versions running a dbt asset-based job was showing me all the different dbt models in “Timed View”, now it is only showing me 1 task, “dbt_project” Is what I am describing above the expected behaviour or am I missing something?

I can't work out how to toggle the time series charts on and off anymore? There used to be an option to do it on the Assets page.

@prha @daniel how AssetMetrializ set the record id of an asset, is it increament 1 on every record or else? Also can you please tell me that how we can set a custom id?

Is there a way to use

**kwargs

in ops? I have a few ops that are rendering Jinja2 template and need to pass information to the templates. It would be nice to be able to generalize the ops so that I can use them in other pipelines and just pass different info into the template, but if I try to pass keyword arguments in I get a

DagsterInvalidDefinitionError

.

dagster.core.errors.DagsterInvalidDefinitionError: Invalid dependencies: solid "update_templates" does not have input "latest_file". Available inputs: ['template_config', 'kwargs', 'start_after']

I’ve tried with an op like this:

@op(
    ins={
        "template_config": In(TemplateConfig),
        "kwargs": In(Optional[dict]),
        "start_after": In(Nothing),
    }
)
def update_temolates(
    context: OpExecutionContext,
    template_config: TemplateConfig,
    **kwargs,
):
    ...

In a graph similar to

@graph
def generate_datasets():
    dataset_path = download_dataset()
    template_config = template_and_destination()
    update_templates(template_config, dataset_path=dataset_path)

How can I yield metadata via context.log? Or the context object fullstop? I was under the impression that this was now possible.

Is there a good way to display less than the full name/path of a nested op in Dagit? For example, the displayed names for

outer_graph.inner_graph.op_1

and

outer_graph.inner_graph.op_2

are truncated based on length to keep the display less messy in Dagit to

outer_graph.inne…

and

outer_graph.inne…

, which becomes very visually unclear with lots of nesting, aliases, or DynamicOutputs.

Hi team, we are looking to migrate Dagit & scheduler from 0.12 to 0.14 and have some questions that hope you could help with: 1. Are there version specific tasks for 0.14 as in https://docs.dagster.io/migration? 2. From how-to-migrate-your-instance, if anything goes wrong, do we just restore postgresql and revert deployment back to prev version?

After upgrading to 0.14.1 we've started to see errors like the below frequently when browsing Dagit (especially the new

instance/overview

page) Is this a know issue; or something to do with our specific set of dependencies?

AttributeError: 'Select' object has no attribute 'subquery'
Traceback (most recent call last):
  ...snip...
  File "/Users/dlaing/workspace/tanzu-dm/.venv/lib/python3.7/site-packages/dagster/core/storage/runs/sql_run_storage.py", line 312, in _runs_query
    subquery = base_query.subquery()
graphql.error.located_error.GraphQLLocatedError: 'Select' object has no attribute 'subquery'

For 14.1 and software-defined assets I no longer ( by default) observe the execution time being tracked in the asset`s detail view. Where can I get it back?

Hey everyone, is the following local setup possible: Have Dagster run pyspark job using the spark-on-k8s-operator from google in a Minikube cluster on a schedule? #mydreamsetup

👋 I’ve tried searching the documentation, source code, and also here in Slack, but came up empty. Does anyone have an example of how to just configure the Dagit and daemon logs to use the

json_console_logger

instead of the

colored_console_logger

. I assume this has got to be a one line addition to a configuration file somewhere (

dagster.yaml

?) but I’m not finding any references as to how exactly to make such a change. The end goal is to be able to route stdout/stderr from these processes into Datadog and have them parsed sensibly, particularly to support multi-line logs (e.g. backtraces).

Hi, I just upgraded 0.13.10 --> 0.14.1, along with

dagster instance migrate

. Seeing a strange bug: • the Assets page is completely blank, no matter which projects I select • but navigating from a job/pipeline run to the asset materialization events shows the assets do exist

With the renewed software defined assets and dbt integration, what’s the best way to schedule a dbt run? Do I still create a job that triggers the CLI, or do I do something else?

Is anyone running Dagster on k8s and successfully passing envs to kube Job? I'm trying to pass down AWS creds from helm chart with no success. I see the creds in the pod, but when the job is launched from the UI, the envs aren't set in the kube Job 😔 . Code is for extra info.

resource "helm_release" "dagster" {
  ...

  set {
    name  = "dagster-user-deployments.deployments[0].envSecrets[0].name"
    value = kubernetes_secret.dagster-aws-secret.metadata[0].name
    type  = "string"
  }
}

resource "kubernetes_secret" "dagster-aws-secret" {
  metadata {
    name      = "dagster-aws-secret"
    namespace = var.namespace
  }

  data = {
    AWS_ACCESS_KEY_ID     = var.AWS_ACCESS_KEY_ID
    AWS_SECRET_ACCESS_KEY = var.AWS_SECRET_ACCESS_KEY
  }
}

Cheers! After some poking around, turns out that the creds need to go into

runLauncher

, which makes sense, modifying the terraform file as this:

set {
    name  = "runLauncher.config.k8sRunLauncher.envSecrets[0].name"
    value = kubernetes_secret.dagster-aws-secret.metadata[0].name
    type  = "string"
  }

Seams to do the trick.

Hey guys, I just updated to 0.14.1 for dagster and dagit. When i try to run dagit I get this error:

2022-02-23 10:11:05 -0500 - dagit - INFO - Serving dagit on <http://0.0.0.0:3000> in process 15370
WARNING:  You must pass the application as an import string to enable 'reload' or 'workers'.

It starts then stops again. Has anyone faced this issue?

Hi Everyone. Wonrdering if there is a universal slack_on_failure that we can configure at dagit level rather than individual jobs.

Does dagster support any role-based authentication mechanism at the moment? I saw there was a github issue (#2219) created for this 2 years ago but don't see any further movement on that. If its not officially implemented, is there a workaround for this that I can use?

Hi Everyone. We are trying to understand an issue related to sensor where we have a pipeline triggered after completion of another and we have created a sensor for it. But the second pipeline seems to be getting triggered repeatedly and its seems like the trigger isn't cleared. Also if we stop the instance and restart the second pipeline continues where it left off. We were wondering if someone else has noticed this behavior and if there is a routine to wipe/clean once a sensor has been triggered and does what its supposed to.

Hi, I have a op (op A) that is computational intensive and part of a job (job A). I want to reuse the output (a pandas dataframe) of op A in job B without executing op A (and all upstreams ops of op A) when i run job B. I do not understand how to do that in dagster.

After updating our dagster instance to the latest version my databricks jobs are suddenly failing - I can’t really figure out what has changed that is causing this, but the job fails on trying to get files from the stdout directory in my job folder on dbfs, which does not exist.

{ 'error_code': 'RESOURCE_DOES_NOT_EXIST', 'message': 'No file or directory exists on path ' '/tmp/dagster_staging/XXX/XXX/stdout.'}

Hello dagster beginner here 👋 I am wondering how dagster integrates with asyncio. I see you can define an async

op

, but you can't define an async graph? at least I didn't see how I could use

DynamicOutput

together with an async

op

. We also use async clients, For instance using

aiohttp.ClientSession

. This is generally used by an async contextmanager. I want to define this as a resource (I saw it is done this way for non-async clients in the hackernews example). How should I do this in dagster?

Hi Team, is it possible to deploy spark docker image in dagster-helm. In this example

dagster-user-deployments:
  enabled: true
  deployments:
    - name: "k8s-example-user-code-1"
      image:
        repository: "<http://docker.io/dagster/user-code-example|docker.io/dagster/user-code-example>"
        tag: latest
        pullPolicy: Always
      dagsterApiGrpcArgs:
        - "--python-file"
        - "/example_project/example_repo/repo.py"
      port: 3030

Can i use spark image in place of docker.io/dagster/user-code-example

Hello, is it possible to add a logger to a sensor?

AttributeError: 'SensorEvaluationContext' object has no attribute 'log'

I hacked this in my test with

context = dagster.build_sensor_context()
context.log = logging.Logger('test')

but running the sensor in dagit gave the above error 🙂