Are there plans to expose

dagster._check

as an “official” top-level module? After I delved into the source code a couple months ago, I found myself importing it into some of my projects, but it still leaves a sour taste in my mouth to import an API that’s not made public.

Hi all! Dagster Product Designer here 👋🏻 We’re working on building a better way to test sensors and schedules within Dagit. I recorded a short demo of a prototype for this new feature and would appreciate any input/feedback from the community on it. We're still early in the design process and all feedback is welcome 🙏🏻 Please leave any comments/questions right here or within the Loom itself. Thank you!

I've been using Dagster since ~0.5.0 in varying degrees. Recently I completed migrating to the SDA paradigm using a combination of DBT + Graph-Backed Assets. My deployment fell very much into this camp before the migration:

Without software-defined assets, you're often forced to:

Split your pipeline into smaller jobs, which allows for easy maintenance but makes dependency tracking difficult

Now in the SDA paradigm I have much needed observability/dependency tracking across repositories for my deployment. Thank you to the whole Dagster team and community for continuing to build amazing products 🎉 !

In Dagit, if I have a step selected and the UI does any updates, it scrolls me back to the step that I have executed. I would rather it only scroll when I 1st select the step (from the "Executing" list for example), but otherwise not auto-scroll at all. For example, I select one step from "Executing" to see some logs and jump to the part of the DAG that's active, and then I want to scroll around to see some other steps, but first I have to unselect the 1st step.

Just a small out of date thing in the docs: https://docs.dagster.io/integrations/dagstermill contains a screencapture from an old version of dagit (0.8.7).

I've set up my company's data pipelines in Dagster and we like it! I was wondering, though - is there some way to avoid using strings for references, etc? Or at least enabling some kind of strict mode that will fail if, say, you pass

non_argument_deps={"something_that_doesn't_exist"}

?

Is there a good page for reference when writing an IO manager? I have written a few that basically take assets and put them into Postgres, and some of the challenges I've run into are: • What is the best way to pass metadata from an

op

to the

io_manager

? What if it needs to be dynamic, e.g. if the op knows what the name of the output should be (setting metadata on the

Output

does not seem to work) • What is the best way to have an

io_manager

that only really works for assets? (thanks to @owen for the idea of the custom decorator) • What are all of the possible sources of the "name" of the thing: ◦

context.asset_key

? ◦

context.step_key

? ◦ various

names

? • detecting partition type (currently I use

instanceof(context.asset_partitions_def

a bunch, maybe there is a better way) • better understanding how

context.asset_partitions_time_window

• How does all of this work in the context of: ◦ Backfills ◦

@multiasset

◦

SourceAsset

◦

DynamicOutput

• Other things to be aware of that may contain sharp edges for those who are writing

io_manager

s Currently what I do is look at the

dagster

test suite, which is pretty awesome, but sometimes doesn't give the comprehensive overview that I sometimes lack. Does something like this already exist/can it?

Small fix for what appears to be a bug in the docs for Dynamic Graphs • https://github.com/dagster-io/dagster/pull/10902 edit: Another one for a bug in the docs for contributing to the docs (so meta!!) • https://github.com/dagster-io/dagster/pull/10903

1. This page seems to be broken currently: https://docs.dagster.io/concepts/dagit/dagit It's back! 2. Multiple command descriptions are truncated with

...

at the end here: https://docs.dagster.io/_apidocs/cli

Are there any thoughts by the Dagster team to have different layouts on Dagit UI for Data Lineage or Graph representation of jobs ? Lot of times, the graph is not very deep, but it can be really wide and in such case the UI becomes pretty clunky to scroll wide. Scrolling up/down is a bit easier for such scenario. So +1 for configurable layout of UI representation of Dags/Assets/Graphs

This looks to me a bug: I initially have 2 non-subsetting assets under a

multi_asset

asset then I add one more. Therefore there's 2/3 materialized and 1/3 missing. I cannot materialize the missing one due to "partition is partial"

I really like the

asset_reconciliation_sensor

with SLAs, but this makes the lack of configuration options in assets and multi-assets even more relevant than it was before. While we can set the configs on the job level, it’s impossible to configure the assets themselves. Maybe I’m still stuck in dagster anti-patterns, but I’d like to keep my code as static as possible and configure assets with YAML files.

Not sure if this is a bug or if I’m misconfiguring something, but it seems like graph-based

AssetsDefinitions

don’t (can’t?) contain metadata. Inspecting the python object does properly return metadata under the property

_metadata_by_key

but it doesn’t show up on dagit and can’t be accessed through

my_asset.metadata

(or rather, the property doesn’t exist).

AssetsDefinition.from_graph(
    my_graph,
    keys_by_output_name={"result": AssetKey("my_cool_key")},
    metadata_by_output_name={"result": {"column_schema": TableSchema.from_name_type_dict({"some_col": "str"})}},
)

Encountered GraphQL error: Error You hit an unexpected error while fetching data from Dagster.

Operation name: AssetGraphLiveQuery

Message: maximum recursion depth exceeded while calling a Python object

Path: ["assetNodes",2,"projectedLogicalVersion"]

Locations: [{"line":44,"column":3}]

are there plans to introduce memory usage graph per run? that would help a ton in debugging/monitoring huge assets.

It would be nice if the

JobSelector

threw a warning or something if a job didn't exist. We had an instance today where we refactored a repo, and didn't find out for several days that this sensor wasnt' being tripped! (Although maybe freshness policies will fix this 🙂 )

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    request_job=my_job,
    monitored_jobs=[
        JobSelector(
            location_name="my_location",
            repository_name="my_repo",
            job_name="my_job",
        )
    ]
)
def my_run_sensor(context: RunStatusSensorContext):
    yield RunRequest(run_key=str(context.dagster_run.run_id))

Proposal: environment-aware assets When delivering ETL code to production it often has to be tested on real production data (this happens a lot in ML pipelines) before merging the code into

master

. To solve this Dagster Cloud's Branch Deployments can be used, or one can setup a custom CI/CD job manually. In both cases, the job spinns up a new temporary Dagster deployment I will refer to this deployment as Feature Stage. The Problem Usually we don't want to have write access to Production from the FS as it would be unsafe. Thus, Dagster jobs must write data into a separate FS environment, like a temporary S3 bucket. However, this means that we also can't read the production assets. One can of course write a custom IOManager that would somehow pick the read and write environment based on the asset metadata, but it would be extremely complicated to support it. In fact, what I'm trying to do here is to generalize this logic and let Dagster run it instead of the custom user code. So when testing an asset, the user has to materialize it's upstream dependencies the FS environment first. This becomes a huge pain when working with heavy assets like trained machine learning models or a lot of data preprocessing for ML. It may take hours to materialize the upstream assets before the developer can finally test the code he was working on. This has an enormous impact on development speed and productivity. Now imagine adding backfills… Solving this problem would bring a lot of value to developers being affected by it. Proposed Solution 1. Introduce a

env

tag to assets and op outputs. Internally it can be just a special metadata value like

__environment__

. The default would be:

__environment__: default

. This tag can be displayed in Dagit. It has to be saved in Dagser's database when the asset is materialized (included in the materialization event). 2. Allow providing a dictionary of

resource_configs

instead of a single config for every resource. Maybe also a dictionary of resource definitions instead of a single one. If the user hasn't provided a dictionary, wrap the single config into

{"default": config}

. 3. When initializing a resource for the asset, pick the resource config (and possible the resource definition) using the asset's

env

tag. If loading an asset, use the tag recorded in Dagster's database. If writing an asset, use the tag provided by the Dagster deployment. Lets see some examples: 1. "Hey let's not overwrite prod files in CI"

with_resources(
    assets,
    resource_defs={"io_manager": my_io_manager},
    resource_config_by_key={
        "io_manager": {"prod": {"base_dir": "prod"}, "feature_stage": {"base_dir": f"fs-{FS_ID}"}}
    },
)

2. "Hey let's not write into prod from CI"

with_resources(
    assets,
    resource_defs={"io_manager": my_io_manager, "aws_credentials": aws_credentials},
    resource_config_by_key={
        "io_manager": {"prod": {"bucket": "prod"}, "feature_stage": {"bucket": "stage", "base_dir": f"fs-{FS_ID}"}},
        "aws_credentials": {
            "prod": {"AWS_ACCESS_KEY_ID": READONLY_PROD_AWS_ACCESS_KEY_ID},
            "feature_stage": {"AWS_ACCESS_KEY_ID": FS_AWS_ACCESS_KEY_ID},
        },
    },
)

What happens inside the feature stage (FS) Dagster deployment: 1. The production Dagster database is being cloned. The FS Dagster thus has access to all the assets materialized in production, as well as runs history, assets metadata, etc. 2. When materializing a new asset, Dagster will load upstream assets from the production environment. The

env=prod

tag will tell Dagster to use the IOManager and resources that have read access to production. 3. When writing the asset, the

env=feature_stage

tag will tell Dagster to use the FS IOManager and resources, thus materializing the asset in the FS environment. As a result, all the assets produced before the FS deployment are going to be loaded from Production. All the assets produced after the deployment will be loaded from and written to the FS environment. The proposed changes changes are: • non-breaking • very general, users can do a lot of stuff with custom environments • would work with partitions • very small codebase edits, we just have to add a few dictionaries here and there • can be immediately used by Branch Deployments in Dagster Cloud Would love to hear what everybody thinks! It's still not very clear to me where does the

env

tag has to be defined - perhaps in the

repository

decorator? Tagging @sandy and @schrockn for future discussions since I've mentioned The Problem to you guys previously.

I like the new

Definitions

way of defining the code location. What's more, it paves the way for enabling Dagster to promote a default (optional) project structure, if sensors and schedules were able to be loaded from package directories in the same way that assets are. Here's what I'd love to see. 1. a new function called

build_default_definitions

from dagster import Definitions, load_assets_from_package_name, load_sensors_from_pacakge_name, load_schedules_from_pacakge_name, load_jobs_from_pacakge_name

def build_default_definitions(
   asset_package_name=".assets",
   sensor_package_name=".sensors",
   schedules_package_name=".schedules",
   jobs_package_name=".jobs"
): 
   return Definitions(
      assets=load_assets_from_package_name(asset_package_name),
      sensors=load_sensors_from_pacakge_name(sensor_package_name),
      jobs=load_jobs_from_pacakge_name(jobs_package_name),
      schedules=load_schedules_from_pacakge_name(schedules_package_name),
   )

then, in a user's auto-generated project

__init__.py

from dagster import build_default_definitions

def = build_default_definitions()

This would make alternative project structures opt-in: users would never have to worry about whether a sensor, asset or job they added is getting sucked into the code location ; if they put it in the right folder, it just happens.

Is there any further traction on this issue https://github.com/dagster-io/dagster/issues/7943 . Specific use cases where preemptive knowledge of partitions is not available until run. Also used for Fanout kind of processing of Jobs/Assets

Could we get left-to-right graph visualization rather than top-down? With operation names written in a language that is written left-to-right rather than top down, we end up with graphs thare are a bit difficult to see on a screen without zooming in a lot, which elimates the bigger picture. Possibly optionally as a toggle button on the screen?

Workspace GQL API It would be nice if there was a mutation to edit the

workspace.yml

so I could have code-locations spin up and automatically announce themselves to the instance, without having to

rsync

the file from the server, edit it, and

rsync

it back.

I'm liking the new Deployment UI. But, I miss being able to see the image and package_name for my code location right in the UI, now I have to click into the little menu and select view configuration (and then scroll to the right because my Docker image name is too long and it doesn't wrap).

In general I feel dagster features the concept of

asset

but it’s strange schedules do not support native asset but we have to convert them to jobs. Sensors recently added support for asset, shall we do the same for schedules?

Many of my pipelines through the years had to limit concurrency due to limited resources such as ram, bandwidth, singletons and so on. The feature that I am currently missing the most is to limit operator concurrency WITHOUT going via dockers. Is that anywhere in the roadmap?

Interesting bug with asset staleness. I’m noticing after upgrading to 1.1.7 that some assets in the asset graph are showing as stale despite only having run once and having materialized in the right sequence, as I checked the run logs to make sure. The

AssetMaterialization

events were emitted a few miliseconds from each other, but in the correct order. It’s also showing differently in the job view in dagit vs. the asset group view. See both screenshots for comparison. The latter is especially surprising as the source asset was materialized 2 minutes before the first asset in the graph. Inspecting each asset individually, all of them show as stale in their asset pages. No

FreshnessPolicies

are configured.

Dagster having a unified history of the execution time for individual dbt models, across time and across dbt Cloud runs (ad hoc vs. scheduled vs. CD-driven), is starting to become extremely useful as we onboard more dbt users. (42 now).

It would be nice to be able to right-click and open in new tab for the deployment name / gears in the deployments menu. (Works for the "View all deployments" item but not the other deployments or gears.)

being able to scroll back in time in the asset plots would be super cool

Custom IO managers and metadata: Would be nice to be able to read the metadata produced by

handle_output()

in the

load_input()

method. In my attempts, using

add_output_metadata()

does work as expected – the event data in asset catalog displays the specified keys – but the keys are not available when reading

upstream_ouput

property of the context in

load_input()

. Modified example from the doc:

class DataframeTableIOManagerWithMetadata(IOManager):
    def handle_output(self, context, obj):
        table_name = context.name
        write_dataframe_to_table(name=table_name, dataframe=obj)

        context.add_output_metadata({"num_rows": len(obj), "table_name": table_name})

    def load_input(self, context):
        table_name = context.upstream_output.name
        
        num_rows = context.upstream_output.metadata["num_rows"]  # <-- That'd be great 

        return read_dataframe_from_table(name=table_name)

If not using the approach suggested above, how else can I read the metadata in the

load_input()

method?

For me, the live view of a running job never shows completion even though the job completes (and shows success in upper left despite last steps still show as not completed). dagit 1.1.6. If I go to the runs page and click on the run the page reloads and show the correct timing.