Hi, I am using dagster-shell solid to create a python virtual environment and install packages into it . Is there a way, I can see the running log of the package installation?

dagster-daemon run on Windows ? It is required to set DAGSTER_HOME variable on windows to run dagester-daemon but it fails with: _dagster.core.errors.DagsterHomeNotSetError: The environment variable $DAGSTER_HOME is not set._ In powershell I set this variable and the folder already exists

Set-Variable -Name DAGSTER_HOME -Value C:\Users\myuser\dagster_home

in the hacker news example, the sensor that run dbt after download searches the run log but describes it as not a recommended pattern (https://github.com/dagster-io/dagster/blob/d5e799c3044c883f583e56a4b4f3791d9937fbe[…]les/hacker_news/hacker_news/sensors/download_finished_sensor.py) - what would be recommended? I’m in a similar situation where I want to run dbt based on how other runs have gone

Guys, is there a way to trigger pipeline B from pipeline A, or to order their execution ay other way?

nevermind , i found the best way to pass configuration , should be in the documentation is there a way to update the wiki to showcase a configured resource?

is there a limit to how long a sensor can take to finish yielding run requests? I got this error:

grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
	status = StatusCode.DEADLINE_EXCEEDED
	details = "Deadline Exceeded"
	debug_error_string = "{"created":"@1634063600.070690200","description":"Error received from peer unix:/tmp/tmps5h2tkhh","file":"src/core/lib/surface/call.cc","file_line":1069,"grpc_message":"Deadline Exceeded","grpc_status":4}"
>

Hi… many thanks in advance… not really sure what I’m doing wrong (but probably something super obvious)… I’m getting this error in

dagit

when I try to run a pipeline (I can see the pipeline I just can’t run it):

dagster.core.errors.DagsterInvariantViolationError: repo not found at module scope in file /opt/dagster/app/repository.py.

  File "/usr/local/lib/python3.7/site-packages/dagster/grpc/impl.py", line 75, in core_execute_run
    recon_pipeline.get_definition()
  File "/usr/local/lib/python3.7/site-packages/dagster/core/definitions/reconstructable.py", line 110, in get_definition
    defn = self.repository.get_definition().get_pipeline(self.pipeline_name)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/definitions/reconstructable.py", line 46, in get_definition
    return repository_def_from_pointer(self.pointer)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/definitions/reconstructable.py", line 518, in repository_def_from_pointer
    target = def_from_pointer(pointer)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/definitions/reconstructable.py", line 460, in def_from_pointer
    target = pointer.load_target()
  File "/usr/local/lib/python3.7/site-packages/dagster/core/code_pointer.py", line 233, in load_target
    name=self.fn_name, file=self.python_file

For reference… I have 4 separate repositories. They are all configured in my

workspace.yaml

:

load_from:
  # Each entry here corresponds to a service in the docker-compose file that exposes pipelines.
  - grpc_server:
      host: dagster_pipelines
      port: 4000
      location_name: "dagster_pipelines"
  - grpc_server:
      host: dagster_experiment
      port: 4001
      location_name: "dagster_experiment"
  - grpc_server:
      host: dagster_dbt
      port: 4002
      location_name: "dagster_dbt"
  - grpc_server:
      host: dagster_data_app
      port: 4003
      location_name: "dagster_data_app"

I’m trying to organize these repositories as dagster projects. Each project has its own associated dockerized image. This is an example of how the files are organized:

├── projects
│   ├── dbt
│   │   ├── dbt
│   │   │   ├── pipelines
│   │   │   ├── schedules
│   │   │   ├── sensors
│   │   │   └── solids
│   │   │   └── repository.py

I’m able to shell into the container and run pipelines. However, in dagit I’m getting the above error. (Perhaps is related to my docker setup?) Thanks again.

I have 10 in-progress runs that have been status:STARTED for nearly five hours, and 170 queued. These jobs typically run in under a minute when I run them a few at a time. What’s a good way to get out of this situation? I’m using the QueuedRunCoordinator and the DefaultRunLauncher and run/event/schedule storage using the Postgres classes, and local logs/storage otherwise. I am running on a two-node setup, one running dagit and the other running dagster-daemon. There are no errors or anything. It seems like the system is just…stuck. all of the jobs are two-op graphs, and the first op is finished, so the jobs aren’t “initializing”. they have finished the first step successfully, which loads a remote file to pass to the next step, which will validate the data and write it to a database. the input to the second op has been loaded using a PickledObjectFilesystemIOManager.

Hello, I have two questions I could find out the answer: 1. Can you actually pass as parameters serialized data between solids? 2. Is there any way to connect pipelines between them?

Hi , My pipeline is successfully executed, but the Dagit UI doesn't show it. What could be the possible reason?

Hello all, I haven't been able to wrap my head around the concept of resources and looking to understand if what I'm trying to do is possible or can be achieved better in a different way. I have a helper class for AWS S3 functions like below -

from dagster import resource, solid, ModeDefinition, pipeline
from dagster_aws.s3 import s3_resource

class HelperAwsS3:
    def __init__(self, s3_resource):
        self.s3_resource = s3_resource

    def s3_list_bucket(self, bucket, prefix):
        return self.s3_resource.list_objects_v2(
            Bucket=bucket,
            Prefix=prefix
        )

    def s3_download_file(self, bucket, file, local_path):
        self.s3_resource.meta.client.download_file(
            Bucket=bucket,
            Key=file,
            Filename=local_path
        )

    def s3_upload_file(self, bucket, file, local_path):
        self.s3_resource.meta.client.upload_file(
            Bucket=bucket,
            Key=file,
            Filename=local_path
        )

The _s3_resource_ is actually _dagster_aws.s3.s3_resource_ which will help me connect to AWS using my local aws credenitals. I am not sure how to pass the _s3_resource_ to the HelperAwsS3 when I make the call in the @resource section below.

@resource
def connection_helper_aws_s3_resource(context):
    return HelperAwsS3()

Any pointers please? Or am I doing it all wrong and it needs doing in a different way? Thanks for your help.

there is the s3 resource already available in the dagster-aws library @Gayathri Chakravarthy

you can use is but constructing the solid like so:

@solid(required_resource_keys={'s3'})
def my_solid(_context):
   ...

and with a pipeline defined and configured as follows:

from dagster_aws.s3 import s3_resource
MODE_TEST = ModeDefinition(name="test", resource_defs={
    "s3": s3_resource
})
@pipeline(mode_defs=[MODE_TEXT])
def my_pipeline():
   my_solid()

you can also configure the s3 resource with a default configuration if you want as follows: just create this mode test instead:

MODE_TEST = ModeDefinition(name="test", resource_defs={
    "s3": s3_resource.configured({"region_name":"us-east-1"})
})

I'm struggling with

UserWarning: Module ... was resolved using the working directory. The ability to load uninstalled modules from the working directory is deprecated and will be removed in a future release. Please use the python-file based load arguments or install ... to your python environment.

errors when using

execute_pipeline()

as shown below:

result = execute_pipeline(
        pipeline=reconstructable(my_pipeline),
        preset="prod",
        instance=DagsterInstance.get(),
    )

Is there a way to

use the python-file based load arguments

when using execute_pipeline() / reconstructable() ?

Hi All. I'm new to Dagster. I deployed Dagster in AWS account using example from

dagster/examples/deploy_ecs

and I'm trying to run the example pipeline, but I'm getting the following error:

dagster.core.errors.DagsterInvalidInvocationError: Too many input arguments were provided for solid 'solid'. This may be because an argument was provided for the context parameter, but no context parameter was defined for the solid.

Which seems strange, since there are literally zero arguments provided. Here's the code of the pipeline:

import time

import dagster


@dagster.solid
def solid():
    time.sleep(30)
    return True


@dagster.pipeline
def pipeline():
    solid()


@dagster.repository
def repository():
    return [pipeline]

Thank you!

hey All, when running a partial backfill we are getting:

dagster.core.errors.DagsterExecutionStepNotFoundError: Can not build subset plan from unknown step: create_full_taker_answers_table

same pipeline runs fine (even partially) if we start in the playground. we cannot see any errors on dagit, celery and user deployment pods. any ideas how to find the problem?

Hello, in regards to this limitation on databricks_pyspark_step_launcher not being able to specify AWS specific attributes for a new cluster, I found a workaround, which is to use an existing cluster with AWS instance profile set up. However, the step launcher config still requires “storage” attribute so I created a placeholder for Databricks secret scope and secret with dummy value. This seems to be working okay since the AWS resource access is managed by the instance profile with the existing cluster, but I wanted to check if there are any concerns with this approach. Where are different places this storage configuration is needed?

how do you guys template sql?

if i have say a template string file, how do i load it correctly into a dagster resource ?

do i just put the sql templates on s3 and load them from there?

Hi team, I’m a bit confused by the interoperability of

local_file_manager

and

s3_file_manager

. I understand they’re designed to use the same API so you can just switch modes, but the former uses a

local_file_handle

and the latter a

s3_file_handle

. So my code still needs some kind of checking to differentiate/determine which one to create or am I missing something here?

Hello, I've ran into a bug with the S3 IO manager. We have a solid that yields multiple outputs. The Outputs have keys along the lines of

output_id

and

output

. The first yield is

output_id

and second yield

output

. The bug comes up when the IO manager tries to upload

output

. The boto3 call

self.s3.list_objects_v2(Bucket=self.bucket, Prefix=key)

will see the prefix

output

and remove

output_id

then upload

output

. The io manager should check for matching ids instead of trusting and removing everything the s3 client lists in the bucket with said prefix.

@solid(
output_defs=[
        OutputDefinition(
            name="clustering_id",
            is_required=True,
            description="versioned clustering identifier",
            dagster_type=ClusteringId,
        ),
        OutputDefinition(
            name="clustering",
            is_required=True,
            description="versioned clustering object",
            dagster_type=Clustering,
        ),
        OutputDefinition(
            name="nmf_clustering",
            is_required=False,
            description="rich NMF clustering object",
            dagster_type=NmfClustering,
        ),
    ]
)
def foo():
    yield Output(bar, "output_id") # Does not exist in the s3 bucket 
    yield Output(baz, "output") # Does exist in the s3 bucket

Hey. Is it possible to restart a daemon through the Dagster UI? Or would I need to restart my Container?

Hello, I am new to programming and I am having problems with the sensors. I get this error in the daemon and I don't know how to fix it: I can't find any tutorial regarding this issue. Thanks!!!

Is it possible to specify an io_root_manager at the graph level in the new API? This would be very useful instead of having to specify it at the op level

Hey! Silly question. I have Dagit, dagster-daemon and repo grpc server running in AWS as ECS services. I'm able to run a pipeline from Dagit UI, but is there a way to trigger run of the existing pipeline in this environment from my local machine using either CLI or API? I was looking through the documentation but it looks like it mostly considers running everything locally

Hey folks, new to dagster. I have set the

run_launcher:      
  module: dagster_celery_k8s
  class: CeleryK8sRunLauncher
  config:
    dagster_home:
      env: DAGSTER_HOME
    instance_config_map:
      env: DAGSTER_K8S_INSTANCE_CONFIG_MAP
    postgres_password_secret:
      env: DAGSTER_K8S_PG_PASSWORD_SECRET
    broker: "<pyamqp://test:test@dagster-rabbitmq:5672//>"
    backend: "rpc://"

The pipeline deploys, but when executing it I see this error

dagster.core.errors.DagsterInvariantViolationError: celery-k8s execution configuration must be present in the run config to use the CeleryK8sRunLauncher. Note: You may also be seeing this error because you are using the configured API. Using configured with the celery-k8s executor is not supported at this time, and all executor config must be directly in the run config without using configured.

Good morning, I am new to dagster and I can't get the sensor daemon to work. My steps to follow:

pip3 install dagster

pip3 install dagit

dagster new-project PROJECT_NAME

cd PROJECT_NAME

pip3 install --editable .

dagit

dagster-daemon run

And I get this error (attached picture): In dagster-daemon documentation they don't give much info.... Thanks!!!