I might be reading this wrong, but is this just broken? Using Dagit 1.0.7

Does anyone else experience the problem with “live” event logs? The problem is I stopped receiving the live updates from pipelines, and only see changes after page refresh. Websocket connection is normal, but I don’t see any messages flowing. I use Dagit 1.0.15 behind Nginx, will try to downgrade and check again.

Hi, I have been trying to get dagster setup with an M1 mac for the last 3 days and have had no success so far. Initially I was getting an error installing grpcio 1.43.0 with python 3.11 and with python3.10 grpcio 1.43.0 installs but when I try to run dagit I get the below error. Any ideas?

ImportError: dlopen [...] lib/python3.10/site-packages/grpc/_cython/cygrpc.cpython-310-darwin.so' (mach-o file, but is an incompatible architecture (have (x86_64), need (arm64e)))

It’s still very difficult to get dagster and dagit working on an M1 mac, much more so than airflow. I’d suggest improving the experience on M1 mac to better compete with airflow

any plans to enable the timeline view to look further back than 24hrs? it's a really great view for reviewing runs and particularly for reviewing failed runs - I usually check it multiple times a day to make sure our flakier jobs are running successfully. but on Monday I can only see back to Sunday, or if I'm out for a day or two I can't see the whole timeline of what I missed. the "# jobs failed" view can be helpful, but the timeline is much more effective for getting a high-level view of what happened and when

hey! What are the scale limits for DynamicOut? Let’s say I have a dataframe with 100,000 rows and I need to do a decently expensive op on each row of that dataframe. Two options are to have an op that goes through all the rows and does the calculation. Or you can yield a DynamicOut per row, but then that’ll generate 100,000 ops. I’m assuming that ’ll break dagster? Is there a limit to how many outputs a DynamicOut op should be?

Having an issue with large logs. I'm logging CURLs at the end of my pipe which includes the data being submitted. It seems its to big and the logs get corrupted and I get GraphQL error. Perhaps some sort of guard that truncates the logs when they get too big would be good? I'm getting errors and can't see the reason as the logs are knocked out

Hi, I first got this error in this thread and I couldn't find a fix for it. I recently ran into the same error and I believe I know where the error is coming from. When we we use

load_assets_from_dbt_manifest

to select nodes from

dbt

, exposures are excluded. In

dagster_dbt.asset_defs

here, only nodes, sources and metrics are selected and then passed to the dbt core's

selector

here. The thing is that the selector function's _is_graph_member expects the manifest to have exposures, otherwise, it will try to find the exposure keys from the nodes, which of course will lead to a keyError,

def _is_graph_member(self, unique_id: UniqueId) -> bool:
        if unique_id in self.manifest.sources:
            source = self.manifest.sources[unique_id]
            return source.config.enabled
        elif unique_id in self.manifest.exposures:
            return True
        elif unique_id in self.manifest.metrics:
            metric = self.manifest.metrics[unique_id]
            return metric.config.enabled
        node = self.manifest.nodes[unique_id]
        return not node.empty and node.config.enabled

I think adding exposures to this list here will make the

load_assets_from_dbt_manifest

not fail when a project uses exposures.

manifest = Manifest(
        # dbt expects dataclasses that can be accessed with dot notation, not bare dictionaries
        nodes={unique_id: _DictShim(info) for unique_id, info in manifest_json["nodes"].items()},  # type: ignore
        sources={
            unique_id: _DictShim(info) for unique_id, info in manifest_json["sources"].items()  # type: ignore
        },
        metrics={
            unique_id: _DictShim(info) for unique_id, info in manifest_json["metrics"].items()  # type: ignore
        },
      exposures={ 
            unique_id: _DictShim(info) for unique_id, info in manifest_json["exposures"].items()  # type: ignore
        } # This line is missing
    )

I can submit a PR if needed.

Hi! 1.0.16 i can't see any more all my schedules in one place? 🫣~.~ upd: sorry,

/overview/schedules

is still good for that. witch one of them is active? maybe instead of count of all schedules in repo, show `active`/`all` info?

Hi, potential bug 🐛 found in the

dagit

UI

dagit==1.0.16
dagster==1.0.16

The visual DAG doesn’t match the declared code when it comes to the visual aspect of

start_after

. It seems the default value of

ins=

within the package defined

@op

takes precedence of what is overridden in the defined

@job

.

from dagster_dbt import dbt_run_op             #dagster bundled library
from dagster_stitch import stitch_sync_op.     #custom
from dagster_tableau import tableau_refresh_op #custom

@job(
    description="Ingest data, DBT transform, do Tableau Refresh.",
    resource_defs={
        "stitch": stitch_resource_all,
        "dbt": dbt_all_models_config,
        "tableau": tableau_resource_all,
    },
)
def workflow_mvp():
    tableau_refresh_op_mvp(
        start_after=dbt_run_op(
            start_after=[
                stitch_sync_op1(),
                stitch_sync_op2(),
            ]
        )
    )

In this instance I would expect the visual DAG to show: 1.

dbt_run_op

should start_after

sync_op1

and

sync_op2

. 2.

tableau_refresh

should start_after

dbt_rup_op

It should be noted that the “Info” pane on the right of the Graph for the respective ops shows the correct values/relationships I believe the visual DAG drawn is using the

ins={"start_after": In(Nothing)}

default declared in the

@op

in the provided

dagster-dbt

library (https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-dbt/dagster_dbt/ops.py#L8) and not the overriding data provided in the

@job

definition. This makes the visual DAG an incorrect reflection of what we have declared in our code, which I would say is a bug sadpanda

One thing I’ve struggled with, and have had some coworkers struggle with when trying to onboard them to dagster is all the ways to express config. We have: • Resource Config ( EG. when we define the configuration of a resource def) • Config of resources (AKA ‘required resources’) • Op config • Run config (which seems to be pipeline config, something thats not really exposed much in 1.xx) • Job Config • input/output config • config via code vs. via yaml • config mappings • sensor configs • more, I may have missed some. • Things that can be configured with @configured, .configured vs. others that cant (even if it doesn’t always make sense intuitively). This isn’t to say this is all bad, obviously being able to configure all these different things in different ways gives a lot of power, but its also hard to know: • With all these options, which should you use? • How to differentiate them (EG: Many times I’ve gotten a a config type to resolve to None, when I thought I’d configured it. Some sort of holistic documentation/guide on configuration may be a good idea, as well as some best practices. I’m still relatively new to the dagster project as a whole so I imagine this becomes second nature after using it for a bit longer. Just thought I’d point out a part of the learning curve that I and the team I work with have found too be a little steeper than expected. Thanks for reading 🙂

Hi there, feature request: It would be nice to have the

Explode graphs

button in dagit for an AssetGroup that was created from a

@graph

using

AssetsDefinition.from_graph

. It’s only possible for jobs as of now.

it feels like we should be able to terminate runs even if they are in STARTING status - when I launch runs using the EcsRunLauncher it takes about two minutes to launch a ECS task, and often I realize I messed up configuration or for some reason need to terminate the job. but then I have to wait around for a couple minutes just to terminate the job.

Since AssetSensors create a dependency between the watched asset and the job that will be run, wouldn't it be nice for Dagit to be able to display this dependency graph? In general, including sensor and schedules in the graph displayed by Dagit would provide a more complete "overview" of a job. I find myself to have to create this graph manually to explain to my team a full data flow I've just created. I don't know if this is already planned or not. Here's an example where the second job is triggered by an

AssetSensor

on the update of the asset

table

.

Flagging this bug which seems to be getting some attention: https://github.com/dagster-io/dagster/issues/10148

Would it be possible to include the root run when filtering the dagit

runs

UI based off of the

root_run_id

tag?

hey friends, cross-posting https://dagster.slack.com/archives/C03SMUY0SFQ/p1668129323199809 looking for users to try out our new secrets UI before it goes GA. Reach out to me on DM if you’re interested.

Why the name "assets"? Isn't a better name "blob"?

I was playing the community edition of Manim (the animation library behind 3blue1brown) and came across this: https://docs.manim.community/en/stable/guides/deep_dive.html I’d been using this library for a while, but never fully understood what was actually happening. Now I’ve read that, I can understand why random bugs I was running in to were popping up. Probably quite a large undertaking, but it could be nice to have a blog post or something which really delved in to what’s going on under the hood of Dagster.

Currently, we are using the

celery_worker

and

celery_executor

with

default_run_launcher

. One problem is that the

User Code GRPC Server

usually runs out of memory. As I check, seems like the

User Code GRPC Server

has to load all the pipeline submitting from the daemon, but somehow it is very memory intensive. I try to run with 3 jobs run, a graph is about 4 nodes x 5 dynamic partition mapping, it costs about 1.1gb of memory (the memory when initializing the API Server is 250mb). Which means around 300mb is allocated for each pipeline. Also when the User Code GRPC Server runs out of memory, all the tasks submitting cannot get to the queues, causing all the running/starting jobs to hang, the only solution is to terminate and restart all. I wonder if we can handle the OOM gracefully to address this problem?

I don’t know if it’s just me, but one thing (easy) that would be nice to change, is that when you log in to dagster in the browser the dagster icon changes colour from purple to nearly white. I really find it hard to find, when I come back to use it from elsewhere 😅. Please could it either change to another colour that’s a bit brighter or stay purple?

Hey team! Can we get some more alert event types? There are some things that aren't job-related that can interrupt pipelines, such as this one

dagster._core.scheduler.scheduler.DagsterSchedulerError: Unable to reach the user code server for schedule datagun_clever_schedule. Schedule will resume execution once the server is available.
  File "/dagster/dagster/_scheduler/scheduler.py", line 505, in launch_scheduled_runs_for_schedule_iterator
    raise DagsterSchedulerError(

The Dagster tutorials feel rather...incomplete and rushed. They tell you how to define assets and jobs, but end there. There's no explanation of what a repository is, what a Dagster project is, or what a resource is. There's also not really a high-level tutorial that takes you through what a Dagster project is at a high level or how these things fit together. There's the 'fully-featured project' guide, but it's not detailed enough and rushes through things too much (in my opinion). Sure, everything is described in extensive detail on the Concepts pages, but that's the problem: there's too much detail on the Concepts pages. Not to mention that the first concept is "Software-Defined Assets" (which the tutorial did cover) but that page then references all sorts of other concepts (repositories, resources, configs) that I never learned about in the tutorial, so how am I supposed to make sense of any of it? What is the intended next step for a new user after they finish the tutorials section? Jump to the concepts page, where you're bombarded with other concepts you have no understanding of? Go to the fully featured project and try and cobble together how the pieces fit together by looking at the code? See my comment in the thread for suggestions on what I think might be a better set up.

Hey everyone 👋 I am currently building my first data pipeline in Dagster in which I want to process data on EMR using PySpark as part of the pipeline. I am looking into the PySpark integration with Dagster. Based on the docs it seems like the recommendation here is to write you PySpark code in Dagster, then submit the Ops as steps to EMR. On one hand this seems pretty cool especially because it is easier to track data lineage (if I understand correctly) but on the other I am concerned that suddenly your transformation code becomes deeply coupled with Dagster, and migrating it might be difficult if you need to do so for any reason. The other option is to use Dagster primarily as an orchestrator and write your code in a separate repo which you can then run on EMR, and trigger the cluster creation and step deployment through Dagster. The second option seems less risky to me. I'm coming from using Airflow so maybe I am looking at this from the wrong lens but I wanted to see what others think about this. With regards to the second option, let's assume that you're using Delta Lake. Would it be possible to create an asset which points to an S3 bucket that reads the manifest file and also the metadata from Delta Lake (or from a metadata store such as Glue on AWS) so that you can still track data lineage without having to write your PySpark code in Dagster itself? P.S. This is more of a discussion rather than feedback, I'm just curious 🙂

UI changes and logos on the integrations are just a beautiful beautiful touch !

Jenkins has a neat little feature in its cron syntax: the

H

value, that would be nice to have for Dagster schedule cron definitions.

To allow periodically scheduled tasks to produce even load on the system, the symbol

H

(for “hash”) should be used wherever possible. For example, using

0 0 * * *

for a dozen daily jobs will cause a large spike at midnight. In contrast, using

H H * * *

would still execute each job once a day, but not all at the same time, better using limited resources.

The

H

symbol can be used with a range. For example,

H H(0-7) * * *

means some time between 12:00 AM (midnight) to 7:59 AM. You can also use step intervals with

H

, with or without ranges.

The

H

symbol can be thought of as a random value over a range, but it actually is a hash of the job name, not a random function, so that the value remains stable for any given project. https://www.jenkins.io/doc/book/pipeline/syntax/#cron-syntax

The idea of using a hash of the job name is very simple and ingenuous, but not necessarily great at spreading the load. In a second stage, maybe Dagster could have something more advanced that would actually look for the times of least activity inside the allowed time range. This ideal schedule time could be set by the first run and kept for subsequent runs, or it could be recomputed every time.

Could we please get support for spaces in the name of various nodes? I can use snake case but our stakeholders would prefer spaces. alternatively possibly auto replace the underscores with spaces... (configurable possibly?)

Hello all. We’re looking to simplify Dagster and remove redundant/unnecessary concepts. To that end we’re introducing a non-breaking change to largely eliminate Workspace and Repository from the system. We’d like to get feedback from the community on this change and to that end I’ve posted a github discussion here: https://github.com/dagster-io/dagster/discussions/10772 We’re looking to move quickly on this and move our examples and tutorials starting late next week (Dec 8th). Thank you!

Just dropping by to say the freshness policy stuff is very cool, kudos dagsir

So i need an Ops semaphore. We use a 3rd party API that is rate limited and if we go over the max concurrent requests then we get shaped way down. We're working up a semaphore API which uses a DB to track which Op has access to the API across different jobs/pipelines, so only one job can query the API at once. We currently limit all the concurrent jobs to 1 using tags but there are slow parts before and after mutex that take time too. I was thinking that it would be cleaner if there was some sort of semaphore argument to an Op to limit access to something. i.e. @op(semaphore="someAPI") which would ensure only one op with that semaphore could run at once across the all dagster instance. I guess this would be a built in API with the daemon process? Since all jobs are sharing a DB anyway wouldn't be that much setup and would be a unique offering from Dagster (i'm guessing)?