hey, just started using dagster and im loving it. currently im using regular typing for all op signatures and wanted to know if theres any types that i have to use the

dagster

version to make them work

Hej! Is there a recommended way to run a multiprocess pipeline locally via the Python API? I’ve found it quite handy to be able to do that to more closely mirror the deployed jobs when testing, and it doesn’t seem to be documented. After a bit of hunting, I got the following construction to work, but it doesn’t appear to be documented, so may not be recommended? Context: being able to run the pipeline via

execute_pipeline()

means I can stay within the VSCode context and use the debugger.

execute_pipeline(reconstructable(my_pipeline),
                     instance=DagsterInstance.get(),
                     run_config=some_config,
                     mode='some_mode')

blob wavereverse Dynamic Partitions Whilst trying to have a 'live' list of jobs/configs I finally found the dynamic partitioned config which seems to mostly do what I want - provides a live / lightweight view of all available runs, and can automatically provide the appropriate config for starting any run. However, I would like to use tags for concurrency limiting (and probably filtering and all that good stuff), and after poking around in the partitions code I can't see a way to adjust the tags for the run. Are there any workarounds or alternative approaches? I think a variation on this was covered earlier, but the suggested workaround I think would only work for a static partition? And even then, the tags couldn't differ for different partitions (e.g. different day of week &tc)? Am I right in my understanding that there's a gap in functionality where runs may have different tags (including e.g. the partition keys), or as generated by a sensor's RunRequest, but the Partition APIs mostly operate on 'run_config' which is a level below a

Job

. Perhaps Partition APIs need to emit RunRequest style objects too, or something...

I’m trying to figure out how to pass a parameter to an

op

within a

job

. I’m trying to follow this one https://docs.dagster.io/concepts/io-management/unconnected-inputs#loading-a-built-in-dagster-type-from-config But when I run dagit it still require me to add configs. Actually, what I’m looking for is a way to have an

op

that takes a parameter (a string), and then pass that parameter to the

op

from a

job

.

I have a job with three independent ops feeding into one final op. I'd like to have the first three ops running concurrently when the job starts, but despite setting the

max_concurrent_runs

to 25, tagging all the ops with "my_key" and setting the tag concurrency limit:

tag_concurrency_limits:
  - key: "my_key"
    limit: 10

When the job runs, it only ever allows two of the ops to start running concurrently, only starting the third one when one of the first two has finished. Is there some other setting I'm overlooking?

Hi Team, My name in Chris Sarakasidis - Data Engineer, Sainsburys. We use dagster (v 0.11.11) on a daily basis to orchestrate our pipelines. Yesterday we cancelled a production pipeline from the UI (see image attached below), but unfortunately it didn’t terminate the job. The job continued and completed causing various issues. This wasn’t the first time it happened, so we are wondering if this is a known bug that was fixed in subsequent versions or there is something else happening that we should be aware of.

Hey y'all - this may be a pretty specific question, but hoping someone out there has figured this out or has an idea of the direction I should take I'm attempting to run a series of SQL statements through a configured Redshift resource (just using the bare dagster_aws.redshift.redshift_resource and configuring it from the launchpad for now). The problem I'm running into is that VACUUM statements can't be run in a transaction block. Using psycopg2, I would just set the connection's isolation level, but I can't figure out how to do that using this resource since I don't think it exposes the underlying connection object. Should I be creating my own custom resource for this?

Hi! Can somebody tell my how can I add this little tags to ops in right bottom corner like "dbt", "sql", "ipynb" in Dagit UI like in this photo? ;) https://mobile.twitter.com/olivierdupuis/status/1410698532894433284/photo/1. I cannot find any information about this feature and I think it helps visualy organise ops in graph.

Hi! I'm a little bit confused about how to run parts of your job, but not all of it. Lets say that I start working in Dagit and I have a really slow job. My pipeline looks like this:

really_slow_op ----> normal_op

So lets say I build this job. Great! Then I need to add a new op that depends on normal_op

really_slow_op -----> normal_op ----> new_op

I want to just run new_op so I don't have to waste a lot of time running really_slow_op. I've tried doing this in the launchpad, and I think I have to do something with the input config. Are there any docs referencing the recommended way to do stuff like this? I could only find the one about reexecuting from an error. Error in dagit, I think i'm misunderstanding what <selector> should be:

Hi, I'm experimenting with Dagster with a use case of creating an hourly run that passes a start_time and end_time into an API call. I've done the hourly schedules using an @hourly_partitioned_config which generates an hourly start time. Am I correct in thinking that this method doesn't expose an end_time and that the logic to generate the end time has to be done in the code of the op. By adding 1 hour to the start time for example. Many thanks!

Hey, our Dagit deployment just crashed throwing the following error, any ideas? We run on 0.12.13

2021-11-04T18:48:51.332707043Z/usr/local/lib/python3.7/site-packages/dagster/core/storage/runs/sql_run_storage.py:179: SAWarning: implicitly coercing SELECT object to scalar subquery; please use the .scalar_subquery() method to produce a scalar subquery.

Hi! I'm trying to migrate to 0.13. How do you set a solid/op retry policy now at the job level? I think

<http://GraphDefinition.to|GraphDefinition.to>_job

doesn't have the retry policy parameter 🤔

Hey, I'm trying to hook my input up to a hard coded string like so:

ops:
  available_capacity:
    inputs:
      flowgates_path:
        value: "/Users/andychen/dev/nira-data-pipeline/pipeline/pipeline/pjm/ops/raw/flowgates/tmp/flowgates.csv"

I think I've done the typing correctly as shown in the image, but I'm getting an error:

dagster.core.errors.DagsterInvalidConfigError: Error in config for job
Error 1: Received unexpected config entry "inputs" at path root:ops:available_capacity. Expected: "{ config?: Any outputs?: [{ result?: { json: { path: String } pickle: { path: String } } }] }".

  File "/Users/andychen/dev/nira-data-pipeline/.venv/lib/python3.9/site-packages/dagster/grpc/impl.py", line 351, in get_external_execution_plan_snapshot
    create_execution_plan(
  File "/Users/andychen/dev/nira-data-pipeline/.venv/lib/python3.9/site-packages/dagster/core/execution/api.py", line 742, in create_execution_plan
    resolved_run_config = ResolvedRunConfig.build(pipeline_def, run_config, mode=mode)
  File "/Users/andychen/dev/nira-data-pipeline/.venv/lib/python3.9/site-packages/dagster/core/system_config/objects.py", line 152, in build
    raise DagsterInvalidConfigError(

any ideas? thanks in advance!

I’m running in AWS with

EcsRunLauncher

and the

QueuedRunCoordinator

with

max_concurrent_runs

set to

100

I have 10 runs in progress and 45 in the queue, but I don’t understand why. This ran perfectly fine in stage with all 58 runs happening concurrently. Any ideas on what’s going on here?

Using pylance in VSCode, and it surfaces type errors on

ins

and

out

:

Argument of type "dict[str, ~_In]" cannot be assigned to parameter "ins" of type "Dict[str, In] | None" in function "op"
  Type "dict[str, ~_In]" cannot be assigned to type "Dict[str, In] | None"

Is there anyway to handle this elegantly, or best to just ignore?

My team ran into a strange bug when launching backfills today. We have some pipelines that are built out of a yaml file by a pipeline factory. Much like in this example. We also have other factories that generate

Run Config Functions

and

Date Partition Sets

for each of these pipelines. Launching partition runs from the playground works as expected, but when running a backfill, all launched runs get the config of the last partition selected for the backfill. Any idea why?

I’m having trouble with Dagster not recognizing DAGSTER_HOME. The directory exists and there’s no writing limitations that I’m aware of. I’m using WSL2 on a Windows10 machine:

Hey there, first of all, thanks for the fantastic tool, my team is starting to explore it and we are loving it! I'm trying to define a

@run_failure_sensor

that reacts relaunching the job it monitors if it fails. Job looks just simply look like this:

@job
def my_job():
    my_op()

Then I define the failure sensor and the repo:

@run_failure_sensor(job_selection=[my_job])
def on_failure_sensor(context: RunFailureSensorContext):
    yield RunRequest(run_key=None, run_config={})

@repository
def my_repo():
    return [my_job, on_failure_sensor]

Problem is that in dagit, sensor says "*Sensor does not target a pipeline*". That being said, if the job being monitored (

my_job

) fails, the sensor will process the fail (log message like

Sensor "on_failure" processed failure of run 032083e3-6bea-4c21-a7d6-85f79d5aa050.

), but won't rerun the job or execute any other code inside the

run_failure_sensor

function. Thanks in advance for your help!

Hi awesome folks! Trying to follow the documentation here: https://docs.dagster.io/integrations/dagstermill#notebooks-as-ops When I try to run the notebook I get the attached error. Not sure where to go from here. What information would I need to provide for proper troubleshooting?

Hello there ! We're in the process of evaluating dagster to move our data pipelines, but we're facing the following issue. Besides that, we have general questions regarding conf and env vars We have this python file

from dagster_snowflake import snowflake_resource

@op(required_resource_keys={'snowflake_resource'})
def get_one(context):
    context.resources.snowflake_resource.execute_query('SELECT 1')

@graph
def say_hello():
    get_one()

say_hello_job = say_hello.to_job( resource_defs={'snowflake_resource': snowflake_resource})


jobs = [say_hello_job]

The thing is that we can't run the job because it's missing some conf (see screenshot) But we did runned dagit with a SNOWFLAKE_USER env var, see below our script to start dagit

export DAGSTER_HOME=/Users/flo/projects/xxx/xxx/dagster/dagster_home
export snowflake_user="TEST"
dagit -h 0.0.0.0 -p 3000

Are we missing something here ? We were expecting that the resource would default the conf from env vars. But more generally speaking, here are our questions, and we struggle to find the appropriate answer in the docs. • What is the way in dagster to configure jobs from vars/files ? We will have 50+ jobs and it seems quite cumbersome to have to pass a config to each, looks like a lot of code duplicates • Also, what is the "official" way to have several resources where just one setting change (the compute_WH in our case), is it using

config_from_files ?

• We would like to configure resources at the repo level, not at the job level, is it possible ? Thanks if you can helps us to get that sorted Our install is for now just a local PoC, using python and venv

Hey there, I want to customize the run_id creation for my jobs (specifically make a datetime.now() string) is there a way to achieve this that would work for both the python API and Dagit?

I was able to get dagster cloud agent running in k8s pretty easy, but having some difficulty finding the knobs to turn to get the user-code portion targeted properly at k8s. I can see the docker setup instructions, but I'm bit lost how to get that wired to my Azure ACR or where to configure k8s.

Exception: Your agent's configuration cannot load locations that specify a Docker image. Either update your location to not include an image, or change the user_code_launcher field in your agent's dagster.yaml file to a launcher that can load Docker images.

Hey everyone, I am having issues trying to get formatted python logs to write to standard out, but I am experiencing some issues. I am using the DockerRunLauncher, however despite my python logger being configured properly as a managed python logger and all logging statements showing up in the dagit UI, it does not seem like any of the logging statements are being written to standard out inside the Docker container spawned for the process. Has anyone experienced an issue similar to this using the DockerRunLauncher, or have an example of how to write to standard out within the process container?

Hello, I am trying to get an AssetMaterialization test running based on the website, but am running into the following after using:

from dagster import job, op, Output

Hello, I have been using dagster-dbt in my projects and it has worked very well so far. dbt has a relatively new command called build. Are there plans to add this command to dagster-dbt in the near future?

I found a broken link on the documentation On this page https://docs.dagster.io/integrations/great-expectations the link for "dagster-ge" does not render.

Is there a way to define dependencies between jobs?

What is the prescribed way to run dagit as a production server? As far as I can tell, it's a Flask server, so would using uwsgi and specifying multiple worker threads be the route?