Q2: Is there a publicly available roadmap for what the future of Dagster will be?

Ask: What is the correct way to stub inputs from other jobs / prior runs when configuring jobs? More detail: I'd like to define a job that runs on a subset of ops in a graph. For the sake of this example, let's say there's only one op that's been selected for run. This op that the job will run needs to grab the outputs of an upstream op that is run in a different job. I'm trying to define it like this: job2 = graph.to_job( name="my_downstream_job", op_selection=["my_op"], config={ "ops": { "my_op": { "inputs": { "input_to_this_op": {"pickle": { "path": "path_to/my_intermediary/in_cloud_storage" # this is produced by the upstream job } } } }, } However, I keep getting this error:

Received unexpected config entry "inputs" at path root:my_op. Expected: "{ config?: Any outputs?:...

---- Situations where we run into needing to do this: The upstream job runs ops that define the train, test, and eval datasets for a supervised learning task. The downstream job runs ops that engineer features and train models . The upstream job and downstream job don't change (or run) at the same frequency. The upstream job that defines the learning task and datasplits runs infrequently. The downstream job runs frequently as the ML engineer iterates on the model itself, and multiple runs should re-use the outputs of a single run of the upstream job.

Hi all, I'm new to Dagster. In our organization we used Luigi to manage some data ingestion and processing pipelines, but we are looking for a newer tool and I'm trying to understand if Dagster is the right one. It has a lot of cool feature, I went through the documentation and did some test, but I'm blocked with a problem, maybe you can help me. Suppose that I have an asset that represents a file on a filesystem. This file may be produced by my @asset function to generate in Dagster. Or it may be already there because, for example, it has been produced by my previous pipeline in Luigi. How can I prevent Dagster to materialize it again if it already exists? I mean, I really want to skip "materialization" ops in case that the expected output already exists, because it would be overwritten. The closest thing I have found is job memoization, but the doc says it's experimental and it isn't applied to assets. Thank you

Thread: In implementing an IOManager's

load_input

method, what is the right way to check if the upstream and downstream assets share a partition definition?

Hi 👋 https://github.com/metaplane/metaplane-dagster/pull/8 anyone know why I’m getting this error where it’s saying there is no dbt_project.yml file in the directory?

Hey everyone! I wanted to get some insight if it is possible to run jobs inside a kubernetes cluster only inside the user deployment pod using something like the DefaultRunLauncher. Currently the issue with the default k8sRunLauncher is that it spawns a new pod each time it runs a job which results in huge overhead especially if many small jobs need to be ran. Since I need to run around 100 smaller jobs each day which are results of partitioning multiple assets with each job lasting about 10s each the overhead of spawning a new pod each time becomes huge. Currently the solution I have in mind is writing a custom launcher that uses already existing pods for each partitioned asset. But I wanted to know if there is a simpler solution out there for this issue before I started working on the custom launcher.

Alternative to Kubernetes At the moment I'm using

docker compose

to orchestrate dagster components in both development and production environments. It's working really well, but I need to plan for a future where I need to scale up run more workers (user codes repositories) on other machines. I'm aware of running dagster on Kubernetes using Helm (https://docs.dagster.io/deployment/guides/kubernetes/deploying-with-helm), but I'm wary of the learning curve. Does anyone have experience getting dagster running on multiple nodes/instances/machines using

docker swarm

?

Hi team! Our dagster event logs table is like 250G now 🙈. Any advice how to properly clean-up? And is there a way to store event logs somewhere else (like s3)?

Hi team, I use this

define_asset_job

, and got the exception below. That was because I used two different partitions definitions for the job and for the asset. What I don't understand is, if the same partitions_def is needed for the job and all of the assets it's materializing, then why do we need to provide the partitions_def when defining the job? Why not implicitly read that partitions_def from the assets? Thanks!

Invariant failed. Description: Assets defined for node 'my_asset' have a partitions_def of Daily, starting 2020-01-01 UTC. End offsetted by 1 partition., but job 'my_job' has non-matching partitions_def of Daily, starting 2020-01-01 UTC..

Hi Team: Today I am getting this error in dagste ui when I am tying to run job:

Operation name: LaunchAssetLoaderQuery

Message: Expected non-None value: None

Path: ["assetNodes",0,"requiredResources"]

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

Hi team, today I do some experiments with

build_asset_reconciliation_sensor

with

dbt_assets

but run into this error:

dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code server

The above exception was caused by the following exception:
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.DEADLINE_EXCEEDED
details = "Deadline Exceeded"
debug_error_string = "{"created":"@1669027129.186794000","description":"Deadline Exceeded","file":"src/core/ext/filters/deadline/deadline_filter.cc","file_line":81,"grpc_status":4}"
>

I run dagit and dagster-daemon locally.

Hi team, I am looking to materialise an op output into 3 different databases? I already have defined 3 different IOManagers, one for each db, but it seems there is no way to specify more than one IOManager for the same output. Which is the correct pattern to implement this? Is there any pattern?

Hi there - I'm having a trouble following the tutorial on my Mac. I'm using Python 3.9.6 and pip 22.3.1. When I enter

pip install dagster

, I get

Defaulting to user installation because normal site-packages is not writeable

. Then when I write a python file with

from dagster import asset

it runs fine. But when I try to use the

dagster

command in Terminal, it shows:

zsh: command not found: dagster

Does anyone have the same issue and resolve it in the past?

Hey, is there a way to configure the executor used by

__ASSET_JOB

which the asset reconciliation sensor and adhoc materializations kick off? What id like to be able to do is have the

__ASSET_JOB

executor be the multiprocessing one (which I think happens by default now) when running dagster locally, but use the k8s job executor when running in the cloud.

Hello! I'm trying to update my dagster installation including dagster-celery integration, but I cannot find dagster-celery library versions matching latest dagster versions. It's still needed to install dagster-celery using pip or is there some other way to use this integration now?

Hello Dagster experts, I have a question: • Is there way to use config value getting from ScheduleDefinition(runtime_config=something) for op_context? ◦ We want to use different values for our op during runtime. So we have yaml file contains multiple schedule configs all using same job with different parameter value. ◦ Currently, we parse yaml to get specific parameters, and assign to SchbeduleDefinition(job=job_a, run_config=parsed_dict) ▪︎ parsed_dict looks like: ◦ args: ◦ value: 1 ◦ Then we want to use it under job_a function: ▪︎ @op(config_schema={“args”: dict}) ▪︎ def _job_a(context): ◦ params = context.op_config[‘args’] ◦ When we run with this, we got the error message saying that: ▪︎ dagster._core.errors.DagsterInvalidConfigError: Error in config for job ▪︎ Error 1: Received unexpected config entry “args” at path root:ops ◦ Seems like runtime_config and op_config don’t share the context information. ◦ Will it be the way to propagate the config value from ShceduleDef level to OP? Thanks for your support!

is it possible for an

asset group

to have a partition?

recently updated k8s helm chart to 1.1.2 -- the daemon is launching the wrong user deployment images for job runs, it's launching old ones. e.g my user deployment is

<http://myregistry.com/user-deployment:v0.2.0|myregistry.com/user-deployment:v0.2.0>

but dagit/daemon is launching the new jobs in

<http://myregistry.com/user-deployment:v0.1.0|myregistry.com/user-deployment:v0.1.0>

. Can I get some pointers on what config may be wrong, or how dagster decides what image to launch for the k8srunlauncher ?

hey! it is possible to have asset partitions defined at only the

job

level? I don’t want to define them at the asset-level, since it is easier (and more DRY) to define them at only the job level. The following code (almost the same as the Parttioned Asset Jobs page) yields no partitions in dagit (v.1.1.2)

from dagster import AssetSelection, HourlyPartitionsDefinition, asset, define_asset_job

hourly_partitions_def = HourlyPartitionsDefinition(start_date="2022-05-31-00:00")


@asset
def asset1():
    ...


@asset
def asset2():
    ...


partitioned_asset_job = define_asset_job(
    name="asset_1_and_2_job",
    selection=AssetSelection.assets(asset1, asset2),
    partitions_def=hourly_partitions_def,
)

Hi, I'm evaluating dagster to see if it fits or needs. Right now we mostly need to run k8s containers/pods (usually non-python) with specific commands/args. I found https://docs.dagster.io/_apidocs/libraries/dagster-k8s#ops , but I couldn't find how I can customize the command I need to run based on input arguments given in my job config... any pointers?

Hey all 🙂 I am trying to load a partitioned asset manually using this

validation_loader = repro.load_asset_value(
    AssetKey("validate_loader"), 
    partition_key='dev_medium'
)

but I get

IsADirectoryError: [Errno 21] Is a directory: '/home/ubuntu/.dagster/storage/validate_loader'

looks like it's going through this code path

upath_io_manager.py:146-149

if not context.has_asset_partitions:
                # we are dealing with a non-partitioned asset
                path = self._get_path(context)
                return self._load_single_input(path, context)

I tried with

1.0.16

and

1.1.2

Hi, guys I am working on pipelineRunLogs subscription on a third party graphql websocket client to keep track of my pipeline run. Everything works just fine, but I cannot get any update on the run status, the status part remains "QUEUED". However, I can get update on step changes and time changes for this run. My dagster version is 0.15.7. Is this my subscription problem, the dagster bug, or my client problem? My subscription is as below

subscription($runId:ID!){
  pipelineRunLogs(runId:$runId){
    ...on PipelineRunLogsSubscriptionSuccess{
      __typename
      run{
        runId
        pipelineName
        status
        stats{
          ...on RunStatsSnapshot{
            enqueuedTime
            launchTime
            startTime
            endTime
            stepsFailed
            stepsSucceeded
          }
          ...on PythonError{
            message
          }
        }
      }
    }
    ...on PipelineRunLogsSubscriptionFailure{
      __typename
      message
      missingRunId
    }
  }
}

Any help matters! Thank you

Hello everyone, I have a problem that I need help with. I'm implementing dagster to sync data from an app to shopify using dagster, using rest api shopify to sync. But I am wanting to write a function test by mock concatenating the api's data with the desired data. I'm having a hard time implementing it right now. We wish everyone help. Thank you everyone

Hi, I'm new to Dagster and I'm slowly trying to use the tools to code my ML pipeline. I'm trying to create an asset whose output is a Tensorflow model. By default, if my function returns a Tensorflow, Dagster will try to picklize it:

@asset
def train():
  model = get_tensorflow_model()
  tf_model.fit()
  
  return model

This fails because TF can't easily be picklized. Though, there is a method to save a TF model inside a file (

tf_model.save("path_to_file")

). Is there a way to tell Dagster to save this file and treat it as the output? While it's nice that Dagster takes care of the serialization, it'd be nice to have an escape hatch in case an asset is not properly supported by Dagster. Thanks in advance!

Has anyone in the community already experimented with a sensor which listens to spark/databricks delta lake changes and triggers a downstream job when a new file has arrived (and is committed in the transaction log)?

I quite like the explanation of `FreshnessPolicy(maximum_lag_minutes=120, cron_schedule="0 2 \* \* \*")``, meaning that by 2AM, c needs to be materialized with data from a and b that is no more than 120 minutes old (i.e. all of yesterday's data).` in https://docs.dagster.io/_modules/dagster/_core/definitions/asset_reconciliation_sensor#build_asset_reconciliation_sensor. However, I wonder if `In order to satisfy this requirement, there must be a materialization for both ``a`` and ``b`` with time >= ``2022-01-02 00:00``. If such a materialization does not exist for one of those assets, the missing asset(s) will be executed on this tick, to help satisfy the constraint imposed` also works well for partitioned assets? I.e. I wonder how the timestamp of materialization corresponds to the partition key and if this considered (i.e. also in case of backfills)

Hi everyone, I'm looking for a doc about sending metrics to Datadog, but could only find the one about sending custom/manual metrics. Does Dagster generate any metrics (e.g: the ones shown in Dagit) by default, which can then be configured to scrape by Datadog/Prometheus,...? I'm trying to have some metrics like: asset materialization time (e.g. the period from when the reader io_manager starts loading the first up-stream asset, till the writer io_manager completed writing out the current asset) - I'm using software-defined assets. Thanks.

Hi All. I'm new to dagster, been playing with it for the last couple months. I have a "Partitioned Non-Asset Job", with it's own schedule and it works good, but the backfill doesn't seem to work as I don't quite understand how to configure the required configs for the job, namely it's resource configs.

dagster._core.errors.DagsterInvalidConfigError: Error in config for job
Error 1: Missing required config entry "resources" at the root. Sample config for missing entry:

I know I missed something, but can't find anywhere, any advise?

Hi there, I had two repositories Repository A Repository B I want to create sensor which will trigger Repository B after successful materialisation of Repository B. Any idea/reference how I can implement it? 🙏

Good morning, just a quick question: Why does

define_asset_job

returns

UnresolvedAssetJobDefinition

? I cant call

.execute_in_process()

to test it. Should I use

resolve

method before calling it?