Question about configured solids. I’m using the dbt module and most of the parameters are set with configuration rather than the input definition. What would be the best way to have another solid generate a value that is supplied to the configuration of a solid? If I wanted to do something like (this doesn’t work)

@solid
def define_dir(input: String) -> String:
  return input

@configured(dbt_cli_run)
def configured_dbt_cli_run(dir_name):
  return {"project-dir": dir_name}

@pipeline
def dbt_pipeline():
  dir = define_dir()
  configured_dbt_cli_run(dir)

I found https://github.com/dagster-io/dagster/discussions/3213 which isn’t quite what I need

i have a pipeline that i'd like to run its main components in all executions, but only run a final solid in production (i.e. only in a specified environment). is it a better practice to do this with a solid that you can configure to be a no-op, or to handle this at the pipeline run level by specifying a solid subset that excludes the final solid?

Quick question: If I want solids on the same "level" to execute in parallel, can I configure something inside Dagster to accomplish this or do I need to implement Celery?

Hey! Sorry to keep asking, but I want an opinion on my use case: Currently I have multiple scrapy crawlers, each one for a different website. I have a solid factory that returns a solid per crawler, and then builds a pipeline with all the generated solids. The thing is, after the first crawler executes, the second one fails as the "reactor" that scrapy uses for its execution cannot be restarted. For this reason I opted to use the "Threads" package, and this solved the problem. The remaining problem is that the crawlers execute, but Dagit fails to detect that they are running. Dagster detects the startup period of the scrapers as their execution time (which is way lower than the normal execution time). What do you think?

Just following up on this question. One other particularly troubling aspect of it is that, when Dagster exits like this, failure hooks are not respected: https://dagster.slack.com/archives/C01U954MEER/p1618852204408800

Hi all. New to Dagster, seeing if it is suited to my task. I'd love some feedback on whether I'm approaching this right. Thanks in advance! I want to create a pipeline to incrementally process a corpus of documents. 1. run a database query that gets a list of PDFs from the source document management system 2. for each PDF in the list, fetch (HTTP get) it to a local directory if it's not already there; delete any local PDFs not in the list 3. for each PDF in the local directory, run a transform to output a JSON file; delete any previous outputs that do not match a PDF 4. do something (reduce?) with the JSON files The transition from 1 to 2 above suggests a dynamic workflow. Should dynamic workflow be used for something where I could have tens of thousands of files, or should it be a static workflow where each solid transforms many documents? I can imagine the solids creating Results related to the PDF and JSON assets. How can I make my pipeline clean up (on subsequent runs) PDF and JSON that have disappeared from the source system?

Hello, I am on 0.11.5 and I see that partitions were the generation was triggered via a schedule, instead of a backfill, do not show up as completed in the matrix under Partition view. Is this expected/known or likely an issue with my implementation?

Hi, I've been trawling the chat histories here, and the API configs, tutorials and documentation, but I can't seem to find an answer to my problem. I'm trying to set up a daily partition set with

from datetime import datetime

from dagster import (
    pipeline,
    solid,
    SolidExecutionContext,
    OutputDefinition,
    InputDefinition,
    String, PartitionSetDefinition, date_partition_range
)


@solid(
    output_defs=[OutputDefinition(name="daily_summary")],
    config_schema={'date': str}
)
def fetch_appointments_for_day(context: SolidExecutionContext):
    ...


@solid(
    input_defs=[InputDefinition("daily_summary")],
    config_schema={'date': str}
)
def ingest_appointments_for_day_to_rdv(context: SolidExecutionContext, daily_summary):
    ...


@pipeline
def client_appointments():
    ingest_appointments_for_day_to_rdv(
        fetch_appointments_for_day()
    )


def runtime_config_for_client_appointment_partition(partition):
    print(type(partition))
    return {'solids': {'fetch_appointments_for_day': {'config': {'date': partition.value}}}}


client_appointment_partition_set = PartitionSetDefinition(
    name="client_appointment_partition_set",
    pipeline_name="client_appointments",
    partition_fn=date_partition_range(
        start=datetime(year=2021, month=4, day=1),
        end=datetime.today(),
        delta_range="days",
        inclusive=True
    ),
    run_config_fn_for_partition=runtime_config_for_client_appointment_partition
)

but the partition tab does not appear in the dagit console. I've tried putting the PartitionSet before and after the pipeline. Is there something obvious that I'm missing?

I can't seem to find a full pipeline with partition example anywhere either Running dagit, version 0.11.5

Does anyone tried setting dagster’s database with microsoft sql server? I know that it is possible since dagster uses sqlalchemy and alembic, but there is no specific documentation. Is there a way to do this?

I have an issue where the sensor won't recognize any new files from S3 so i tried to debug it and found this file which defines the

get_s3_keys

function i use in my sensor. I saw that for every sensor run, it lists all files in a bucket with specific prefix, sort them by

LastModified

, then runs on every key and tries to find the file equal to the

since_key

which in my case it's the

context.last_run_key

. I suspect my problem is that i deleted the files in S3 and that causes this logic to fail because the s3_key will never be equal to the since_key (line 38) . Is that a bug or a feature? how am i suppose to handle this kind of situation? Thanks!

is there a way to just pull a specific named output out of a solid that generates multiple outputs when defining a pipeline? the docs have examples of term expansion, e.g.

a, b = some_solid()

, but i'm not seeing an example of something like

b = some_solid().get_output('b')

, which i prefer for clarity reasons

Hi I am trying to generate airflow dag dagster-airflow scaffold --module-name <..> --pipeline-name <..> --preset prod Error: Invalid set of CLI arguments for loading repository/pipeline.

Hi! Just running into a trouble with sensors example my setup: • dagster 0.11.5 • DAGSTER_HOME is set for Dagit and DagsterDaemon to equal dir • DagsterDaemon is running

export DAGSTER_HOME=~/dagster_home && dagster-daemon run

• code is identical copy from example (except MY_DIRECTORY is set to some dir) • code for pipeline and sensor placed in one file "test_sensor.py" • running dagit with command

export DAGSTER_HOME=~/dagster_home  && dagit -f sensor_test.py

•

run_storage

,

event_log_storage

,

schedule_storage

are set to Postgres instance in

~/dagster_home/dagster.yaml

What I see: deamon log:

SensorDaemon - INFO - Not checking for any runs since no sensors have been started

No schedules or sensors defined

in dagit UI, also

/instance/sensors

web page tells

No sensors found

I can run pipeline successful with custom config in a playground

solids:
  process_file:
    config: {"filename":~/dagster_home}

Please help me debug this, where to look to find the problem?

is there a way to set a preset for a sensor? I see a mode param.

my current workaround is passing e.g.

run_config=preset_dev_env.run_config

to the

RunRequest

inside a function decorated with

@sensor(pipeline_name="my_pipeline", mode="dev")

, but it would be nice to pass e.g.

preset

to the decorator.

Hi, hoping someone can help. • We've setup Dagster Helm K8 deployment on EKS-Fargate-AWS. • It's in a VPC. We added a Client VPN. We've setup all the AWS VPC/VPN networking steps. • We want to keep Dagit private to those with VPN access. It's on a private subnet. Still no access to Dagit, even though we have a working instance with a local IP on the VPC. Is there a concept we are missing for Kubernetes networking? How do companies normally access Dagster through here? https://docs.dagster.io/deployment/guides/kubernetes/deploying-with-helm The deployment was the easiest part... we are stuck on the networking now? Any help is greatly appreciated. Thanks!!!

i'm currently testing a pipeline by running dagit locally, and am seeing some odd behavior that might be a bug: i have hooks bound to several of the solids to send messages on success, and the hooks require the 'slack' resource. however, when i run the pipeline, the logs don't show it initializing the slack resource, and none of the hooks are firing.

Hi, what is the best way to dynamically define solids and adding them to pipeline at runtime? For instance, if I want to create solids from a list retrieved from a DB.

jasono's question above got me wondering. What is the best way or usage pattern in dagster where I can pass in a list of file names or paths where every solid in the pipeline will be given a file path in the list as input then with each subsequent pipeline run, it will loop through the list of file paths. So the structure of the pipeline is the same or not dynamic, but instead, the input to the pipeline is what will be dynamic.

Hi, I’am trying to run a pipeline and after some time (not always the same, sometimes 1.30 minutes other times 10 mins) I get a postgres connection error and the pipeline fails. The pipeline consists of seven solids and each one of them executes a bunch of queries to BigQuery. The whole pipeline on production should take a couple of hours to finish. We don’t have this problem with our other Dagster pipelines which are a lot smaller and shorter in time. Any help is greatly appreciated, thanks!!

Hi All, I’m trying to write a pipeline which would dump the processed data into two separate locations (databases). Wondering what would best practices suggest to achieve this?

Hi all, i'm just at the phase of evaluating whether dagster is a good fit for our project. 90%+ of our processing is on image files which need to be saved to a persistent storage during (and after) a run. Then likely urls/filepaths would be passed between solids rather than the images themselves. Is this a well defined pattern in dagster? From what I can tell, I should to use a

FileManager

but I can't get any further than that. Could someone explain how I would do this or point to an example of a pipeline that passes file handles rather than pickled data? Edit: this would be running on GCP in production so I would be using the

dagster_gcp.gcs_file_manager

but I dont know how to use it

Hi Everyone, We're trying to document our dagster solids using sphinx and none of the functions show up in our API docs. I'm assuming this has something to do with the way the functions are wrapped and that the docstrings are getting dropped somewhere. Is this a known issue? Does anyone know where I could look to find the root cause?

Hi people, this is day #2 of Dagster so please bear with me… I’m getting

Received unexpected config entry "config" at path root:solids:images. Expected: "{ outputs?: [{ result?: { json: { path: String } pickle: { path: String } } }] }".

when setting a .yaml file like

solids:  
  read_xml:
    inputs:    
      path: 
        value: "src/data-in/cumulus.xml"
  camera:
    config: "src/data-out/camera.csv"
  images:
    config: "src/data-out/images.csv"

it looks exactly like the cereal tutorial so I’m confused about where the error is

This message was deleted.

Have some problem with running dagster-celery==0.11.5 as a cli Can’t forward ADDITIONAL_ARGS to celery worker working command:

dagster-celery worker start -A <http://dagster_celery.app|dagster_celery.app>  -y celery_worker.yaml -q queue1 -n worker1

How can I add

--concurrency=10

to that command? try to just append it to the end was not succeed:

dagster-celery worker start -A <http://dagster_celery.app|dagster_celery.app>  -y celery_worker.yaml -q queue1 -n worker1 --concurrency 10
Usage: dagster-celery worker start [OPTIONS] [ADDITIONAL_ARGS]...
Try 'dagster-celery worker start --help' for help.

Error: no such option: --concurrency

I saw here that cli just forwards ADDITIONAL_ARGS to celery command at the end, but I think that when dash or double dash are used - click validation fails (no such option) Also I cant find a tests for ADDITIONAL_ARGS in dagster_celery_tests (was looking for usage examples) Thanks in advance!

i'm writing some tests for some configuration code we put into place, and i'm a bit stuck trying to validate that a

configured

invocation is correctly configuring an object. if i call

configured(foo)({'a': 'b'})

, how can i validate that the

a

value in the solid config has been set to

b

?

if I delete a pipeline run with a given

run_key

is that enough to get the sensor to trigger again (assuming it yields that runrequest again)? (trying to resolve a situation where many pipeline runs that were triggered by a sensor failed -- with bad run_config -- and I want to re-trigger them all with the fixed run_config)

For

great_expectation

integration, is there a particular reason why running a checkpoint is not supported? It looks like it supports running a suite. Just wanted to make sure that I didn’t miss it.