Hello, I've stumbled upon some behaviours in dagster which I think are bugs, but can't be sure. For context, I have a batch ML pipeline, which can be simplified to:

() = solid
[] = OUTPUT

   (load_training_data)         (load_serving_data)
            |                           |
     [TRAINING_DATA]              [SERVING_DATA]
            |                           |
(create_training_features)   (create_serving_features)
            |                           |
    [TRAINING_FEATURES]         [SERVING_FEATURES]
            |                           |
      (train_model)                     |
            |                           |
     [TRAINED_MODEL]                    |
            |                           |
            +-----------+   +-----------+
                        |   |
                    (serve_model)
                          |
                    [PREDICTIONS]
                          |
                (publish_predictions)

Due to restrictions in my work environment, I'm unable to use any of dagster's scheduling capabilities - let's say all we have is cron jobs. I would like to run this pipeline on two separate schedules in different ways: 1. Annual model train, i.e.

execute_pipeline(
        instance=my_instance,
        pipeline=my_pipeline,
        step_selection=["*train_model"],
        run_config={
            "resources": {"io_manager": {"config": SOME_PERSISTENT_IO_MANAGER_CONFIG}},
            "solids": {
                "load_training_data": {"config": "path_to_training_data"},
            },
        },
    )

2. Monthly model "serve" (make predictions), i.e.

reexecute_pipeline(
        instance=instance,
        pipeline=my_pipeline,
        parent_run_id=find_most_recent_training_pipeline_run(instance, ...).run_id,  # Implementation not important
        step_selection=["load_serving_data*"],
        run_config={
            "resources": {"io_manager": {"config": SOME_PERSISTENT_IO_MANAGER_CONFIG}},
            "solids": {
                "load_serving_data": {"config": "path_to_serving_data"},
            },
        },
    )

I think this pattern should allow me to have a single view, in dagit, of this entire model's pipeline, including • every execution of the "training" subset • every execution of the "serving" subset, along with which execution of the "training" subset was its "parent" This should also allow me to: • retrain the model on an ad-hoc basis, where subsequent "serves" use the most recently trained model • easily answer the question for any predictions: "which data trained the model that produced this output"? The problems I'm having are: 1. This second ("serving") execution fails due to missing configuration for solids that I'm not trying to run (e.g. "load_training_data"). Q1: Is it a bug that re-execution with a specific step_selection requires configuration values for solids that won't be executed? 2. When reexecuting via "reexecute_pipeline", there is no indication in the dagit UI that this second ("serving") execution is a derivative of the parent ("training") run. • When re-executing in dagit, the UI indicates this by showing a 1-dimensional 'graph', with one marked as the 'root' • I've been able to work around this by explicitly assignbing the run tags: "dagster/parent_run_id" and "dagster/root_run_id", but this feels brittle. Q2: Is it a bug that these tags are automatically applied when running from dagit, but not when using "reexecute_pipeline"? Apologies for the wall of text, and have to say I'm loving dagster since introducing it at work!

Does dagster have a similar concept to that of a DummyOperator in Airflow?

getting the following when trying to execute a pipeline from dagit:

dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline collective_pipeline
Error 1: Received unexpected config entries "['s3_bucket', 's3_prefix']" at path root:resources:io_manager:config. Expected: "['base_dir']."

my pipeline’s ModeDefinition has a lin in the

resource_defs

that read

"io_manager": s3_pickle_io_manager,

. and I have the run config in the playground set as follows:

resources:
  io_manager:
    config:
      s3_bucket: <bucket>
      s3_prefix: <prefix>

I don’t understand where i’m messing up here ….

Hello! I migrated one of my pipelines to the new 0.12.x "job" abstraction. In my repository, if I am using the dict form, I am getting an error. If I am using the list form, I am not getting an error.

Hi again! The pipeline that I migrated over to 0.12.2, takes a lot longer to execute than pre 0.12.0. It usually completes in 30-40 seconds, but now it takes over 2 minutes. The difference I have found so far is it appears with 0.12.2, by default, the execution runs are using multiprocess execution, instead of in_process execution. I looked at my pre 0.12.0 pipeline version in dagit, it is using the "in_process" execution by default and I can see that there are 2 options available in dagit for the "execution" config in the Playground tab: "multiprocess" and "in_process". But when I look at the 0.12.2 dagit version, only "multiprocess" is available. Is there any way to disable multiprocess execution or switch to "in_process" execution? In my job or schedule definition, I just have:

<http://my_job.to|my_job.to>_job(resource_defs={'io_manager': fs_io_manager")

and my dagster.yaml just has:

telemetry:
  enabled: false

So not sure why only multiprocess is available or perhaps it is being specified somehow that I am not aware of. EDIT: Ahhh, nevermind. I see there is a

executor_def=

parameter in the

to_job()

method. 😆

I am trying out nesting graphs, but getting

KeyError: my_graph1

Example code:

import time
from dagster import graph, op, in_process_executor, repository, ScheduleDefinition


@op
def expensive_task1(context):
    <http://context.log.info|context.log.info>('Performing expensive task1...')
    time.sleep(3)


@op
def expensive_task2(context, my_input):
    <http://context.log.info|context.log.info>('Performing expensive task2...')
    time.sleep(5)


@op
def expensive_task3(context, my_input):
    <http://context.log.info|context.log.info>('Performing expensive task3...')
    time.sleep(5)


@graph
def my_graph1():
    task1_done = expensive_task1()
    task2_done = expensive_task2(task1_done)


@graph
def my_graph2():
    task3_done = expensive_task3()


@graph
def my_graph_final():
    my_graph1()
    my_graph2()


my_job = my_graph_final.to_job()


@repository
def dev_repo():
    return [my_job]

Last one for today 🙂 So looks like

config_schema=

is still available for the @op decorator. Will there be a rename for the word "solid" to "op" in the context object and then for to_job()'s

config=

, "solids" will be renamed to "ops"? For example:

context.solid_config['some_value']

will change to

context.op_config['some_value']

and then

my_job = my_graph.to_job(
    name='My_Job',
    config={
        'solids': {
            'expensive_task1': {'config': {'plant_code': 'Plant_Code_123'}},
            'expensive_task2': {'config': {'plant_code': 'Plant_Code_123'}}
        }
    }
)

will be

my_job = my_graph.to_job(
    name='My_Job',
    config={
        'ops': {
            'expensive_task1': {'config': {'plant_code': 'Plant_Code_123'}},
            'expensive_task2': {'config': {'plant_code': 'Plant_Code_123'}}
        }
    }
)

Correct? Or are we keeping the solid/solids references still for some reason?

Hi guys, quick question I have deployed dagster on a kubernetes cluster on Amazon’s EKS. Now, all the run history etc is stored in a

dagster-postgresql-0

pod which refers to an EBS volume (say A) through a

StatefulSet: default/dagster-postgresql

. Now, when I delete the cluster (i.e. all the resources are cleared up using

eksctl delete cluster

), and when I create a new one + deploy dagster again using

helm

, it creates a new EBS volume (say B) for me. So even though A exists, but dagster now uses B and thus I don’t have my previous run history preserved. So, to make the run history persistent across deletion and creation of the cluster, what do you guys recommend? 1. Is there anyway I can tell helm to use the existing EBS volume (A) instead of the newly created one (B)? 2. Or do you recommend that I use RDS instead of this postgres thing? In case of RDS, there are 2 approaches I could find. a. First, https://docs.dagster.io/deployment/guides/aws#using-rds-for-run-and-event-log-storage i.e. having all this stuff set in a

dagster.yaml

file which I could then supply inside my Docker image. b. OR secondly, using https://docs.dagster.io/deployment/guides/kubernetes/customizing-your-deployment#configuring-an-external-database in the

values.yaml

file. So, can it be done through

1)

? If not, then for RDS, do I need to do

a)

or

b)

or both?

Hi team, I am thinking about the backfill implementations. Currently all my sensors are looking for the latest asset key (generated by upstream deps) for the current partition and create Run Requests if it identifies a new run key which is not the same as the last run key in the cursor. But it looks like while running backfills on the upstream pipelines, these downstream sensors might not pick up the assets and I might have to trigger backfills on downstream pipelines separately. I am currently planning to remove the partition filter in the sensors while calling

events_for_asset_key

, but I am still not sure how to get the partition info from the Asset key to pass it in the run request. The returned

EventLogEntry

does not have the partition info. Is there any other method that I can use?

version 0.12.2: What is the appropriate way to add a custom description to the graph or job in the dagit UI? There doesn't appear to be a

description=

parameter for the

to_job()

method. I tried regardless, but the description didn't changed. I also added a description to the

@graph

decorator and that didn't do it either.

This solid (stripped of business context) is returning the below error. As far as I can tell it's identical in pattern to all the other solids I've written. Any ideas / suggestions? ExpectationResult passes e.g. data is being retrieved successfully from the query.

Solid get_data did not fire outputs {'result'}

Is there a recommended pattern for getting/setting the

dagit_base_url

for the slack hook config without hardcoding it? It would be nice to be able to set it as a metadata property per mode and then dynamically grab it from that.

Is there a way to reset the next expected time that a sensor will execute? e.g., I turned a 1/day sensor on at 8:00am, but maybe now I want it running daily at 8:00pm instead. Turning it off/on doesn't reset the last run time, which it seems to base its next run time off of.

Is there a way to specify the output_config_schema via the configured API? I have an IO manager that I want to specify the schema for that would be easier to just define when the pipeline is defined. Right now I end up doing it in the schedule

Hi all, I am working on a project where in docker I am creating 3 different containers : 1.Dagit UI, 2.my code container(where all my pipelines are listed in repository.py) 3. dagster daemon container My code container has all the pipelines listed in repository, which I am executing using graphql submit_pipeline_execution function. I have a separate Dagster.yaml file as well which indicates all my settings to store event logs in postgres database. Before executing the pipelines using graphql client, we used to execute_pipeline directly and inside solids we have our customised logger functions to store all the activities happening inside the solids. On integrating the graphql client with DAGIT ui, my customised logger functions are not working, seems it is not going inside those functions itself. context.log.info is working properly as I can see the event logs with Run ID captured in postgres database. Is there any way where in I can have a my event logs and all my logging info statements getting logged in my single log file? dagster.yaml: run_storage: module: dagster_postgres.run_storage class: PostgresRunStorage config: postgres_db: hostname: env: DAGSTER_POSTGRES_HOST username: env: DAGSTER_POSTGRES_USER password: env: DAGSTER_POSTGRES_PASSWORD db_name: env: DAGSTER_POSTGRES_DB port: 5432 event_log_storage: module: dagster_postgres.event_log class: PostgresEventLogStorage config: postgres_db: hostname: env: DAGSTER_POSTGRES_HOST username: env: DAGSTER_POSTGRES_USER password: env: DAGSTER_POSTGRES_PASSWORD db_name: env: DAGSTER_POSTGRES_DB port: 5432 My customised log function: log_info_statement(p_in_log_file=l_log_file, p_in_statement='Exiting', p_in_function=l_function)-----calling inside the solid def log_info_statement(p_in_log_file, p_in_statement, p_in_function): if not logger.handlers: handler = logging.FileHandler( l_log_file) formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) logger.info(f'{str(datetime.now())} - {l_function}') logger.info(f'\t{l_statement}\n') Dockerfile of my code container: CMD dagster api grpc -h 0.0.0.0 -p 3001 --working-directory $(pwd) -f ./orchestration_manager/repositories/datamax_etl_repository.py

Hi, can we use spin job pipelines in separate namespaces ?

like per team basis

I'm sure it's very commonly used but I can't work out what the tool is that converts the GitHub dagster docs into web markdown / html. e.g. code within the docstrings being displayed as examples. Could someone let me know? Would be interested in using it internally.

hey, I'm trying out the new core api and interested in the intended usage. context: our pipelines are usually separated into folders with their run_configs and schedules (which we quite like as everything is contained in a single dir) and they are imported to the dagster repo. the problem with this, that we need to add the same modes and tags to every solid and pipeline, this is where

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

could help I believe. is there any way to keep

run_configs

schedules

close to the

graphs/pipelines

and

resources

tags

close to

repos

like:

[<http://g.to|g.to>_job(resources, tags..) for g in graphs]

what's the intended way to organise multiple pipelines + run_configs?

I see some of the dbt solids allow you to pass through a state path (like dbt_cli_seed) - any chance that this will be expanded to the dbt_cli_run solid - or will I need to create my own solid with the flag?

Minor but I'm assuming the description at the top of a pipeline definition is meant to come from the docstring in the pipeline function definition? Can't find a way to generate it.

Guys, I have a Dagit instance which has a workspace pointing to another host which has a gRPC server serving the repo. Whenever I triggered a run from Dagit UI, I can see that the gRPC server spins up new processes to run the pipeline, which is expected. My question is, which party will write execution results to the DB? Dagit or the repo itself? If it’s the later, where should I config the repo to let the repo know the credentials of the DB?

Guys, I am trying to establish a database connection through dagster and currently, I am feeding in database_URL with the Username and Password. I don't want to hardcode that and would like to use a JSON File which has credentials located in an S3 bucket

Hello, I have a dagit pipeline that processes a TIF image and writes the results to a DB table; the results are two columns a string of the combined coordinates and a float number and is about 100'000'000 rows long. The process works perfectly fine on my local machine yet fails when deployed on an EC2 instance. I have increased the RAM on the ec2 instance to be inline with my local machine; 16GB, the number of cores is less (4 on ec2; 12 on local machine). Yet, around the same point when I wish to apply an aggregation to the pandas DF to remove any duplicates, I get the following error regardless of what I do to try remedy any memory issues in the code: - ENGINE_EVENT [DagsterApiServer] Pipeline execution process for 39c9a1ae-5598-4dd1-ba25-cd305f8b4d26 unexpectedly exited with exit code -9. 172257.677 - PIPELINE_FAILURE This pipeline run has been marked as failed from outside the execution context. 172257.699 I have tried everything I can think of in the code to try and overcome this issue, however, nothing appears to help and I can't find any information on what this error code means, what caused it or how I can fix it etc.. If someone could help me out with this, I would greatly appreciate it - I have been stuck on this for a couple days and can't find a resolution to this problem. Thanks and kind regards, Breandán Kerin

Hi, I was looking for documentation on how you can execute a Solid as a Kubernetes Pod We use this functionality in Airflow at the moment for when we have a custom job we’d like to execute. Wrapping this in a container has proven useful (especially when developed by another team) as it gives process isolation from Airflow. Can someone help?

I'm using config schema to inject complex config into my pipelines. I'm wondering if there is a way to compile the schema into a dataclass or NamedTuple? So I get autocomplete and better type inference? say

# Generated from path.to.my.module.__my_config_schema()
@dataclass
class MyConfig():
    field1: str

@solid(
    config_schema=__my_config_schema()
)
def my_solid(
    context: SolidExecutionContext
) -> MyConfig:
    return MyConfig(**context.solid_config)

We re-deployed Dagster from Kubernetes to AWS ECS (with the same Postgres database) and are now getting unexpected GraphQL errors when looking at the “Runs” tab. Could that be related to the fact that both deployments were running in parallel with the same database for a few hours? (See thread for error message)

Is there a way to inject some init code to execute in the run pod before pipeline runs please? We use vault sidecar to store secrets and it needs few seconds to init. I am hoping to set a delay in the pipeline and then add secrets as env vars so that Dagster StringSource configs could use them seamlessly.

Hi all, not sure if asked before (did a Slack search, but maybe I didn't use proper search terms). Purely a learning question... If my test dagster instance uses: • Postgres{Run/EventLog}Storage, • and just the default

io_manager

which db table are the intermediate result outputs of ops/solids stored in (i.e. I am guessing that whatever mechanism allows me to rerun a pipeline/job from the latest successful intermediate step in a given run, is pulling that stored data from somewhere)? Or is it stored as a file/blob in whatever storage backend I configured?

Is there a way to set my write logic via Asset / Resource definitions? e.g. embed the code to update an asset in some form of definition of that asset. Feel like such functionality fits in with the idea of technical implementation agnostic solids, same code within the solid irrespective of whether I'm writing to a MongoDB or Postgres, or do I configure that by providing the same interface / method for each in my class definition of a resource? Basically I feel like the AssetMaterialization / asset_key and actual data write should be more closely coupled.