Hello again blob wave I am using a factory pattern to generate my pipelines (I've also tried the same arrangement with one pipeline and multiple presets). I need this because the pipelines configurations aren't known upfront, but configuration can be determined automatically. In development the pipelines all refresh fine, but I suppose that is dagster daemon doing code hot-reloading, or a byproduct of it using a multiprocessing executor that just starts a new interpreter each time In production using the recommended grpc server pattern

load_from:
  - grpc_server:
      host: executor
      port: 4000

Clicking the "reload repository" buttons on the UI all indicate they are doing something, but the pipelines are never updated. Previously I suspected the 'lazy repository' pattern (I was returning a callable as per docs) but even returning JobDefinition objects it still doesn't work. It requires a full container restart to force a reload:

@repository
def dev_repository():
    return list(pipelines_factory(mode=mode_development))

My questions are: • what am I doing wrong? • how can I have a containerised deployment that supports reloading (manual, or preferably automatically via a 'sensor'?) • or perhaps I've misunderstood the functionality - what is the act of 'reloading repository' for, if not to re-fetch repository content? TIA! 😀 (0.13.0, but I couldn't get it to work prior to that either)

is there a way to programmatically terminate runs? I’m trying to figure out how to build a sensor that will terminate runs if they are taking longer than expected

Hi, I am using user-code-deployments. I want to specify executable path for each repository. My workspace.yaml and helm configurations are in the images attached. Seems like it is not working, it is not using the environment mentioned in the executable_path fields.

What's the simplest way to deploy Dagster in a AWS (least external dependencies to configure)? I'm piloting using Dagster and want to prove to our team that it's relatively simple to get up and running for real, in a way multiple folks can access and where execution can scale. I ran across

dagster-aws up

in the release announcement for 0.6 but when I

pip install dagster-aws

there's no executable, did that functionality go away? I also found someone's Terraform module which doesn't look like it's seen much adoption; and I can follow the Deploying Dagster to AWS docs but it leaves setting up an RDS and an executor as exercises for the reader.

Hi, Is there a way to pass args to shell script using

dagster_shell.create_shell_script_solid

?

We are developing an end of day process in dagster. So far it is working fine and we have done the POC. Now we are looking to parallel process customers but not able to find right documentation. Follow is definition.

in dagster

0.13.1

, is there a way to “populate” multiple config presets in

launchpad

UI? previously we have two configs:

dev

and

prod

(I assume it needs

Mode

), with the new API changes not sure how I should migrate edit: I think I found what I’m looking for https://dagster.slack.com/archives/C01U954MEER/p1635091915068000

I am facing an issue to run the example of static partitions. The issue is coming from the op where it logs continent_name. op_config is None

Hi , values.yaml

- name: "example-user-code-3"

image: repository: "dagsterimage/user-code-example-v37" tag: "latest" pullPolicy: Always dagsterApiGrpcArgs: - "--python-file" - "/home/orchestrator/system/system.py" < EFS mounted path > port: 3030 But i am not able to access repos. getting this error : FileNotFoundError: [Errno 2] No such file or directory: '/home/orchestrator/system/system.py Can we access user-code-deployment repos from efs location if i configured this efs path like above sample code.

Hello, I am part of a team that has a Dagit instance running on dagster 0.12.9 with an associated daemon and we've had a couple instances of the daemon crashing unexpectedly with the error in the attached image. I just wanted to check if this was an error people were familiar with/with a known cause. The machine the instance is running on is a Linux machine with fairly limited memory if that's relevant.

currently upgrading my code to support

0.13.1

found out that using

slack_on_success

and

slack_on_success

yield two behaviors: 1. it is marked as

legacy

in

dagit

2. it displays error

python
dagster.core.errors.DagsterInvalidConfigError: Error in config for pipeline
Error 1: Received unexpected config entry "config" at path root:execution. Expected: "{ multi_or_in_process_executor?: { config?: { in_process?: { marker_to_close?: String retries?: { disabled?: { } enabled?: { } } } multiprocess?: { max_concurrent?: Int retries?: { disabled?: { } enabled?: { } } } } } }".
  File "/usr/local/lib/python3.9/site-packages/dagster/grpc/impl.py", line 342, in get_external_execution_plan_snapshot
    create_execution_plan(
  File "/usr/local/lib/python3.9/site-packages/dagster/core/execution/api.py", line 727, in create_execution_plan
    resolved_run_config = ResolvedRunConfig.build(pipeline_def, run_config, mode=mode)
  File "/usr/local/lib/python3.9/site-packages/dagster/core/system_config/objects.py", line 152, in build
    raise DagsterInvalidConfigError(

but, if I remove slack decorators:

python
# @slack_on_success("#dagster", dagit_base_url=os.environ["DAGIT_BASE_URL"])
# @slack_on_failure("#dagster-fail", dagit_base_url=os.environ["DAGIT_BASE_URL"])
@job(resource_defs=resource_defs, config=prod_config)
def cookbook():
    one = step_one()
    two = step_two()
    step_three(one, two)

it works as normal. in addition, the slack api docs still use legacy API. https://docs.dagster.io/_apidocs/libraries/dagster-slack I assume the migration docs is in the works? blob smile happy eyes

Hi, How to pass parameters to pipelines dynamically so that the parameters can be modified before pipeline execution. In the below example , may i know how to modify the BUCKET_SIZE at runtime.

BUCKET_SIZE = 10
@pipeline(
    mode_defs=[MODE_DEFINITION],
    preset_defs=[PRESET_DEFINITION]
)
def test_data_pipeline():
    some_data = get_some_data()
    for i in range(BUCKET_SIZE):
        process_some_data(i)(some_data)

@solid()
def get_some_data(context):
    <http://context.log.info|context.log.info>(f"Starting !")
    all_files = get_all_files()
    val = np.array_split(all_files, 10)
    return val

def process_some_data(value):
    @solid(
        name=f"process_some_data_{value}",
        tags={
            "value": f"{value}",
        }
    )
    def _process_some_data(context):
        index = context.solid_def.tags['value']
        ...
        ...
    return _process_some_data

Hello all, I am using the external-source feature of dbt, so I have dbt command that looks like this: dbt run-operation stage_external_sources --args 'select: f45_gf' --vars '{"start":"2021-10-05", "ext_full_refresh": "true"}' How do I invoke this via dagster? I don't think I can use context.resources.dbt.run(), am I wrong?

Quick question - How do I make my Dynamic Outputs run in parallel? I fanned out one of the solids but its only running sequentially.

Hello everyone, does anyone have any knowledge on how to debug a pipeline run in Microsoft Visual Studio Code by running dagit through the debugger? I'm able to debug pipelines through unit tests via pytest, but I'm struggling to find any resources related to running dagit through the debugger. I'm aware of the Dagster CLI functionality as well such as dagster debug, but I am not sure how to leverage these tools.

Hi team, is there any way to configure a job with multiple partition sets? I would want to be able to have both monthly and daily partitions on my jobs.

not sure whether a bug or not, but: if I set

config

in

@job

and trigger it via

schedule

, it doesn’t display

configuration

in the

schedule run UI

, although the pipeline seems to be running with the correct config

Hi! Is here a way to persist installed libraries for py env? I’m creating multiple python environments inside dockerfile and then during running jobs, installing required libraries. However, after jobs finish running, libraries aren’t installed anymore for that environment. Using deploy_docker example

Hi Just a quick question. I know we can define dependency between solids in a pipeline, but is it possible in dagster to define dependencies between two different pipelines ??

Hey do you anticipate any issues with starting like a job per second via the graphql api? (Assuming a suitable deploy)

hey guys any way to prevent the execution of a c_solid? I am using conditional outputs and passing them to the composite but I get this error:

raise DagsterInvalidDefinitionError(
E   dagster.core.errors.DagsterInvalidDefinitionError: @composite_solid 'c_solid_create_re_uw_template' has unmapped input 'enable'. Remove it or pass it to the appropriate solid invocation.

Hey all, There is a way to run only one instance of the scheduled job at the same time?

Hi, I want to push logs to cloudwatch and I can do it fine using dagster_aws.cloudwatch.cloudwatch_logger when I execute from the cli and pass in a config .yaml as a parameter as shown in the docs:

from dagster import job, op
from dagster_aws.cloudwatch import cloudwatch_logger

@op
def my_op(context):
    <http://context.log.info|context.log.info>("Hello")
    context.log.debug("World")

@job(logger_defs={'cloudwatch': cloudwatch_logger})
def my_job():
    my_op()

But I can't seem to figure out how to hard-code in the logger config for each job in my docker dagster deployment. Where in the above should I be passing in the config dict for my my job to instantiate the cloudwatch logger with my required settings?. I've tried this, but I get errors, I guess because it's expecting a logger definition and not an actual logger instance:

from dagster_aws.cloudwatch import cloudwatch_logger
from dagster import job, op, build_init_logger_context

cloudwatch_config = {
    "log_level": "INFO",
    "log_group_name": "dagster-test",
    "log_stream_name": "my-job",
    "aws_region": "eu-west-1",
    "aws_access_key_id": "ABCDEFGHIJKLMNOP",
    "aws_secret_access_key": "12345678910"
}

my_job_context = build_init_logger_context(logger_config=cloudwatch_config)

@op
def my_op(context):
    <http://context.log.info|context.log.info>("Hello")
    context.log.debug("World")

@job(logger_defs={'cloudwatch': cloudwatch_logger(my_job_context)})
def my_job():
    my_op()

Any idea what silly oversight I'm making?

Hi everyone, I'm trying to switch my dagit run launcher to use the DockerRunLauncher as opposed to the DefaultRunLauncher, and I am able to pull the proper image for the process run. However, when it goes to start the container for the process, the container is never created and it hangs in "starting" state. Attached is a screenshot of the engine_event below. Has anyone run into this issue before and know some possible solutions? Are there logs that the RunLauncher writes to?

Hi team, I am still on 0.12 and I have a question on failure_hook. I followed solid-hooks to create a function

slack_message_on_failure

annonated with

failure_hook

and annotated my pipeline with

@slack_message_on_failure

. However, the hook doesn’t really get triggered when the pipeline fails. Is this because I am using a databricks spark step launcher and the failure event are STEP_FAILURE and PIPELINE_FAILURE? What’ the right way to set up a hook if pipeline fails please?

Hi team, When we are creating pods using K8sRunLauncher in pipeline execution time. it's showing pod > status : NotReady. But pipeline executing successfully in this k8s pods i checked logs everything looks fine. we are deploying user-code-deployment helm chart with Argocd tool in this tool the k8s pods still in progress state only. How i can solve this pods NotReady status issue.

Improve performance of jobs Hello, I wonder how we can improve jobs performance ? I experience big delay between each ops (10s in local dev mode with SQLite and with UI). I also see difference of 5s duration for 30s jobs between UI and CLI. How do I know where dagster spend most of the time ? I will share further information when I am permitted to do so.

I'm setting up a local Dagster dev environment following the instructions at https://docs.dagster.io/community/contributing and running (hundreds?) of errors similar to the following when running

make dev_install

➤ YN0013: │ @babel/helper-create-class-features-plugin@npm:7.14.0 can't be found in the cache and will be fetched from the remote registry
➤ YN0001: │ Error [ERR_STREAM_PREMATURE_CLOSE]: Premature close
    at new NodeError (node:internal/errors:371:5)
    at PassThrough.onclose (node:internal/streams/end-of-stream:122:30)
    at PassThrough.emit (node:events:390:28)
    at emitCloseNT (node:internal/streams/destroy:145:10)
    at processTicksAndRejections (node:internal/process/task_queues:82:21)

previously I could set

partition offset

in

monthly schedule

in

0.13.1

I couldn’t find the parameters anywhere (

monthly_partitioned_config

). is this by design? (my current workaround would be to subtract one month manually) edit:

end_offset

seems to be the param in the updated API. The docs could probably mention this as well?