Hi team, we want to set tag concurrency limit to set max quota for each user code deployments that are running in our dagster deployment to make sure one of them don't take all the allowed max limit. We also want to apply limits to certain use cases under each of the user code. Does the team recommend any specific structure for this use case? We are currently trying something similar to below

queuedRunCoordinator:
  enabled: true
  config:
    max_concurrent_runs: 25
    tag_concurrency_limits:
      - key: "user-code-1"
        limit: 15
      - key: "user-code-1"
        value: "usecase-1"
        limit: 10
      - key: "user-code-1"
        value: "usecase-2"
        limit: 5

      - key: "user-code-2"
        limit: 15
      - key: "user-code-2"
        value: "usecase-1"
        limit: 10

      - key: "databricks"
        limit: 10

Is there a really simple example project that uses the new core APIs (graph/job/op)? Maybe there’s a development branch somewhere that uses them for the

dagster new-project

jinja templates?

Hello, I'm trying to run the dagster-slack example but I'm getting the following error

AttributeError: 'SlackResource' object has no attribute 'chat_postMessage'

Hello, I'm trying to test this op below which has both a config_schema and a resource key:

@op(
    config_schema={
        "plant_code": str,
    },
    required_resource_keys={"values"},
)
def get_shipping_receiving_data(context) -> Nothing:
    plant_code = context.op_config["plant_code"]
    begin_date = context.resources.values["start_date_yyyymmdd"]
    end_date = context.resources.values["end_date_yyyymmdd"]

    ...

But not sure how to go about it with the latest dagster API. I'm thinking I need to do something along the lines of:

from dagster import build_op_context

def test_get_shipping_receiving_data():
    context = build_op_context(
        resources={
            "values": {
                "config": {
                    "start_date_yyyymmdd": "20210801",
                    "end_date_yyyymmdd": "20210803"
                }
            }
        },
        config={"plant_code": "some_plant_code"}
    )

    result = get_shipping_receiving_data(context)

How to use hooks with new graph API? I am using

execute_in_process

to run the graph and looking for a way to extract out information (error message/exception) in case pipeline fails.

[help/feature request] I've woken up this morning to a collection of Dagster run failures for a particular pipeline/job. I'd like to know whether these are all related to the same root cause. Drilling into each run failure on by one I can see that it's the same solid failing. Is it / would it be possible to: (a) get a list of the failed solids associated with each failed run? (b) get an aggregated list of log (especially error log) messages across pipeline runs?

Hi all dagster mask I am currently facing a blocker in Dagster while trying to use 

build_reconstructable_pipeline

Use case: Create runtime pipelines (ie. set solid_defs and dependencies in PipelineDefinition on runtime) with solids processed using a multi-process executor. Have integrated with spark to process data. I looked up the dagster docs and tried to code it up, but I am facing an errors with it. I have attached the relevant code snippet and the error in this thread. For `build_reconstructable_pipeline`: 1. What should I do to resolve this error and use it for my use-case? 2. Could you please provide an example code for this.

Hi, we have setup Dagster on AWS using ECS. Now we want to use a S3 Bucket as IOManager. The documentation https://docs.dagster.io/deployment/guides/aws that we have to add a yaml block for resources: io_managers to our pipeline config. This is not very well documented, since I don't know which yaml file to put that config in. Do I have to add this block to the dagster.yaml or the workspace.yaml?

Hello! I'm trying to upload a file (CSV) to a S3 bucket, looking into the code there is a solid to handle this task https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-aws/dagster_aws/s3/solids.py

def file_handle_to_s3(context, file_handle):

But and unable to find what a file_handle is or how to create this object instance from a CSV file. Could anyone help me please? @chris @max

Hello Everyone, Is it possible to get a

dagster.op

executed even if some of the input dependencies are not resolved? I tried marking input as

Optional

and

is_required=False

on parent output. Currently the step is getting skipped saying:

Skipping step query_datapoints due to skipped dependencies: ['relative_to_absolute.split_requested_period_meta'].

Hi everyone, What does this error mean and what triggers it? dagster.serdes.ipc.DagsterIPCProtocolError: Process exited with return code 1 while waiting for events

hey guys, the API that triggers a pipeline run will return a UUID to the run, is that correct?

Hello! Is it possible to specify a k8s secret env vars map for celery-k8s?

Hi, seems that the latest version of great-expectations (0.13.x) doesn't work with Dagster, Is there any suggestion? I would like to use the new features of GE, like modular expectations definitions. I just started my development so, I don't want to redo it again in the near future.

Hi team, I have some questions around the solids. Can we allocate number of pods for one specific solid? Also, is there distribute processing of records across pods?

Hello team! I am trying to run a DL model and of course, I got to the point of needing to import a model (file or file directory) to the solid in order to run it. What is the best way to approach this point? reading in the documentation I see that the IO Manager is only recommended for testing. In case of production, what is the best way to store these files (not time-variant) and use them whenever needed?

Dagster team, happy Friday! I’ve run into interesting behavior: in our K8s (with K8sRunLauncher) deployment, when I execute a pipeline using any executor (in process, multiprocess, k8s step), our code’s boto client fails to retrieve S3 credentials from the EC2 metadata server. When running the script locally in the very same run container that Dagster spins up, however, it just works! For context, we are authenticating with S3 via

kiam

(the pods are annotated with the proper IAM role). Is there something going on with the way Dagster executes runs that is causing this? Hope this is interesting-- thanks.

Hi team, I have a pipeline A that does some pre-computation and then waits for an asset from pipeline B before proceeding. Is there a way that I could create a solid in pipeline A to listen to asset materialization events from other pipelines please?

I’m working on migrating my tests to use

build_solid_context

with direct invocation instead of

execute_solid

but I’m running into some inconsistent behavior with configuring the logger. With

execute_solid

I was able to pass this with the config to suppress the DEBUG level being output to console:

{'loggers': {'console': {'config': {'log_level': 'INFO'}}}}

But when I directly invoke the solid with

context = build_solid_context(config={'loggers': {'console': {'config': {'log_level': 'INFO'}}}})
some_solid(context)

The DEBUG level is still being output to the console, is there anyway to suppress that?

On a somewhat related note, is it possible to pass Nothing dependancies with direct invocation? I have the

start_after

defined in the input_defs but calling the solid with

start_after=None

gives:

TypeError: cohort_inputs() got an unexpected keyword argument 'start_after'

But when the start_after arg isn’t included then:

dagster.core.errors.DagsterInvalidInvocationError: No value provided for required input "start_after".

hi team, wondering if you could help with pointers to debug jobs pipelines stuck after “[K8sRunLauncher] Kubernetes run worker job created” please?

Guys, do we have plan to support building pipelines out from solids from different repositories?

Hi all. I have Dagster running on AWS EKS but I get the following the error message:

Operation name: JobMetadataQuery

Message: (psycopg2.OperationalError) server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.

[SQL: SELECT event_logs.id, event_logs.event 
FROM event_logs 
WHERE event_logs.run_id = %(run_id_1)s ORDER BY event_logs.id ASC
 LIMIT ALL OFFSET %(param_1)s]
[parameters: {'run_id_1': '2a23d5cb-681b-4cc8-b493-5d68b75a3687', 'param_1': 0}]
(Background on this error at: <https://sqlalche.me/e/14/e3q8>)

Path: ["pipelineRunsOrError","results",0,"assets"]

Locations: [{"line":60,"column":3}]

Stack Trace:
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executor.py", line 452, in resolve_or_error
    return executor.execute(resolve_fn, source, info, **args)
  File "/usr/local/lib/python3.7/site-packages/graphql/execution/executors/sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/schema/pipelines/pipeline.py", line 270, in resolve_assets
    return get_assets_for_run_id(graphene_info, self.run_id)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/implementation/fetch_assets.py", line 59, in get_assets_for_run_id
    records = graphene_info.context.instance.all_logs(run_id)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 1013, in all_logs
    return self._event_storage.get_logs_for_run(run_id, of_type=of_type)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 234, in get_logs_for_run
    events_by_id = self.get_logs_for_run_by_log_id(run_id, cursor, of_type)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 201, in get_logs_for_run_by_log_id
    results = conn.execute(query).fetchall()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1263, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 324, in _execute_on_connection
    self, multiparams, params, execution_options
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1462, in _execute_clauseelement
    cache_hit=cache_hit,
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1815, in _execute_context
    e, statement, parameters, cursor, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1996, in _handle_dbapi_exception
    sqlalchemy_exception, with_traceback=exc_info[2], from_=e
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1772, in _execute_context
    cursor, statement, parameters, context
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 717, in do_execute
    cursor.execute(statement, parameters)

When I try to run a pipeline it does start but does update the job status very slowly (sometimes hours after the end of it) if not at all. I guess it is somewhat related to the above error and postgresql. Any help would be much appreciated!

Is there a way to enable a pipeline schedule by default? Noticed that all schedule is in disabled state after creation. I don’t see relevant parameter in https://docs.dagster.io/_apidocs/schedules-sensors#schedules

What's the best way to make a sensor update the cursor only when a pipeline ran successfully?

Hi Folks, is there any example that shows me how to use the S3FileManager? I want to write multiple files from one solid to S3. I already got the IO manager working, but I think thats not what Iam looking for, since it only allows to write one file. Cheers, Matt

Hi team, what’s a good way to reset sensor cursor please? Previous run failed and I’d like to have sensor trigger pipeline again after fix. Should I use graphQL?

Is there a way to retrieve what runConfigData is missing from a config schema via GraphQL Api? P.S Not the PipelineConfigValidationInvalid error message.

Hi guys, I am using dagster daemon on a network behind a proxy, both http_proxy and https_proxy env variable are set. When I start the dagster daemon everything runs as expected: I have a sensor that fires when needed, but the command line logs an error every second. The error is: E0831 123758.808000000 151232 src/core/ext/filters/client_channel/http_proxy.cc:82] 'https' scheme not supported in proxy URI . Is there a way to avoid it or at least to avoid it to be printed? Thanks for your help.

Hi team, what would be the best way to add a timeout to a solid? I'm using the default RunLauncher, QueuedRunCoordinator, and the default IO Manager. I'm kinda lost on how the solid exists (it's a thread, it's a process etc) and IDK if I can use Unix Signals or something...