Good afternoon all! The standard dagster message level is DEBUG (which is completely reasonable) and I see documentation about how to change the message level at run time through the run_config. Is there a way to set this for the length of a python session? Like through the logging package? Or to at least match the default logger level? Thanks!

Hi there. Running Dagster on K8 using AWS EKS Fargate. I get OOMKilled for playground and scheduled runs. How to increase container memory request / limits? Is there a YAML setting to help ensure this? The run pods must have a very tiny limit, as it is small data sizes. It is only provisioning pods of size 0.25vCPU 0.5GB RAM. I am using the k8sRunLauncher, and it doesn't seem to have a 'resources' property like the others e.g. celery launcher. Thanks

Hi, quick question. I wanted to execute a pipeline with dagster cli and then check it on dagit UI on my local. I executed the pipeline with

dagster pipeline launch -p hello_pipeline -w user-deployment/deploy_k8s/example_project/workspace.yaml -l example_repo2

and it worked on my terminal. But the run doesn’t reflect the output result on the dagit UI. Isn’t the command

dagter run

not supposed to show the result on the UI? and if so is there any way to do this with the cli?

Hi Everyone, How to create mysql connection pool which can be accessed in pipelines?

Hey Guys ! Quick question. How can i loop over Solids ? for example, Solid number 1 and 2 will run multiple times with the output of solid 3.

Hey team, I am trying to run a scheduled pipeline over a partition set of some 500 items and I keep getting this error:

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

  File "/usr/local/lib/python3.7/site-packages/dagster/scheduler/scheduler.py", line 212, in launch_scheduled_runs_for_schedule
    debug_crash_flags,
  File "/usr/local/lib/python3.7/site-packages/dagster/scheduler/scheduler.py", line 258, in _schedule_runs_at_time
    scheduled_execution_time=schedule_time,
  File "/usr/local/lib/python3.7/site-packages/dagster/core/host_representation/repository_location.py", line 687, in get_external_schedule_execution_data
    scheduled_execution_time,
  File "/usr/local/lib/python3.7/site-packages/dagster/api/snapshot_schedule.py", line 55, in sync_get_external_schedule_execution_data_grpc
    else None,
  File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 272, in external_schedule_execution
    external_schedule_execution_args
  File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 97, in _streaming_query
    yield from response_stream
  File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 426, in __next__
    return self._next()
  File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 826, in _next
    raise self

I can manually run the backfill over this partition set though 😞

Any pointers how I should mount volumes on job pods with K8sRunLauncher?

Hi team! I was wondering if you could help me with this logic. I need to reuse the same logic I had in this solid but using different resource (I have several different postgres databases so the resource is the same but the name and credentials different). I am not sure how to parametrize this because the resource is called in the code with the name of the resource). I mean I would like to avoid have the same logic twice just because the resource changes.

It will be awesome if we could access to a specific resource using a string .

Hi team another question. How I can configured the config_schema in order that one parameter could be optional. For example, here I am expecting string but None is also possible.

A question on parallel execution. This example demonstrates how to write a parallel pipeline using the Celery executor. Could the same be accomplished with the K8sRunLauncher?

Hey, I have a question about the celery docker executor. When I configure it for a pipeline and try to run it through the dagit

Playground

tab, I get the error saying that the

celery-docker

is not a valid config entry under the

execution

path. How can I configure this executor? Am I missing a particular step? I have the

dagster_celery_docker

requirement installed in the environment that runs dagit with the

DefaultRunLauncher

.

Hey team, I am evaluating Dagster to be used within my org. I am currently exploring of having separate user code and dagit deployment by following this guide (https://docs.dagster.io/master/deployment/guides/kubernetes/customizing-your-deployment#separately-deploying-dagster-infrastructure-and-user-code) . One question, is there a way to add workspace or repository without having to configure the dagit deployment? It seems that dagit deployment need to know all repository servers beforehand.

Hi, I got this error

dagster.core.errors.DagsterInvalidInvocationError: Compute function of solid 'test' has context argument, but no context was provided when invoking.

when I

dagster pipeline execute -f sample/repo.py  -p source_graph

and my repo.py is

from dagster import op, graph, repository
from dagster_shell import create_shell_command_solid

@op
def source_op():
    command = "echo 'hi all'"
    a = create_shell_command_solid(command, 'test')
    a()

@graph
def source_graph():
    source_op()

@repository
def repo():
    return [source_graph]

I’m not sure where I’m doing wrong here. Could someone help me out on this?

Hi, does anybody know what could be causing user code deployment to crash loop based on readiness probe? Here is the error: grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "failed to connect to all addresses" debug_error_string = "{"created":"@1628578337.829694221","description":"Failed to pick subchannel","file":"src/core/ext/filters/client_channel/client_channel.cc","file_line":3008,"referenced_errors":[{"created":"@1628578337.829689392","description":"failed to connect to all addresses","file":"src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc","file_line":397,"grpc_status":14}]}" > File "/usr/local/lib/python3.7/site-packages/dagster/core/workspace/context.py", line 494, in _load_location location = self._create_location_from_origin(origin) File "/usr/local/lib/python3.7/site-packages/dagster/core/workspace/context.py", line 413, in _create_location_from_origin return origin.create_location() File "/usr/local/lib/python3.7/site-packages/dagster/core/host_representation/origin.py", line 263, in create_location return GrpcServerRepositoryLocation(self) File "/usr/local/lib/python3.7/site-packages/dagster/core/host_representation/repository_location.py", line 480, in init list_repositories_response = sync_list_repositories_grpc(self.client) File "/usr/local/lib/python3.7/site-packages/dagster/api/list_repositories.py", line 14, in sync_list_repositories_grpc deserialize_json_to_dagster_namedtuple(api_client.list_repositories()), File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 163, in list_repositories res = self._query("ListRepositories", api_pb2.ListRepositoriesRequest) File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 110, in _query response = getattr(stub, method)(request_type(**kwargs), timeout=timeout) File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 946, in call return _end_unary_response_blocking(state, call, False, None) File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking raise _InactiveRpcError(state) But, for the moments that is hasn't crash looped*, I can actually access the repository through Dagit and run things.* So, it seems to be a readiness probe issue?

Hi, I'm currently playing with Dagster at the moment. Not sure if this is a good use case but would you recommend Dagster for stitching a bunch of Python functions containing some API calls, dataframe methods and string analysis? These functions are short running (ms to few secs) and there maybe 20-50 functions that can be called in different sequences so the pipelines will have to be created dynamically. Also another thing is does it make sense to just deploy Dagster as a python module to check and execute pipelines if there is no need for the dagit and dagster daemon to keep track of runs? Thanks in advance.

The default directory structure that results from using the s3_pickle_io_manager only uses the solid name (and some ids?), would it not make a lot of sense to default to having the pipeline name somewhere in the path? More importantly how would I configure this? Do I need to overwrite a method of s3_pickle_io_manager?

Hi, is there a way to support obeservability on dagster ?

Hey guys, I think i found a bug using ‘time.sleep’ module in Solids.

@solid

def time_test():

time.sleep(1)

it’s running infinitely (supposed to finish executing after 1 second)

Hi! Currently trying out dagster and beginning to love it very much! But is there any way to use swarm cluster to spawn RunContainers in it?😊 Or, maybe configure docker with TCP socket for dagster - not with UNIX?

Forgive me if this somewhere in the documentation, but I'm having trouble finding it.... is there some way to set a default timeout on a pipeline? For context, here's my use case: I run Dagit in a single container which I deploy via dokku and attach to a hosted Postgres DB. Sometimes when I deploy changes, if i'm not careful, I forget to stop any currently running jobs. This can lead to zombie jobs which are no longer running but are stuck in a "Running" state. Since I use a

QueuedRunCoordinator

, this can, over time, lead to no new jobs launching if there are enough of these zombie jobs. Is there a way I can configure, on a per-pipeline basis, a way of auto-cancelling jobs once they've run for a certain time period?

I'm using Dagster with PySpark. Using execute_pipeline returns the

result

object and I'm able to view the output with

result.output_for_solid("solid_name").show()

, etc. Is there a way to access a similar interface for a previous run? For instance, I run the same pipeline via Dagit, it writes the intermediate and final DataFrames to their respective locations. I want to load the parameters of the pipeline by

run_id

, as an example parameter, and access the intermediate and final DataFrames (stored in Parquet fwiw) via a similar

result.output_for_solid

interface - is that possible?

Hello again, we’re introducing type checking in our project and I was wondering if there’s a way to get more meaningful stack traces from failing built-in type checks. I see you can create custom exceptions for user-declared constraints using

ColumnConstraintViolationException

, but is there a way to use this with builtin types? Like printing out which column failed the test and the offending rows, for instance

Hey, could someone shed a bit of light on the

Snapshot ID

of a pipeline run? Is it generated from a hash of the pipeline file? And further, is it possible to retrieve and run pipelines of a certain snapshot?

Hey. I am relatively new to Dagster. For my use case, I have solids predefined but the pipeline is to be constructed at runtime which may consist of subset of solids. What's the best way to do this?

Hello, is it possible to get the solid's alias name from inside the solid, maybe using

context

argument? This is to check which solid is currently invoked in a pipeline with more than 1 same solid but different alias.

I am having intermittent issue when trying to execute example pipeline using GCS IOManager

An unexpected exception was thrown. Please file an issue.
google.api_core.exceptions.Forbidden: 403 POST <https://storage.googleapis.com/upload/storage/v1/b/my-bucket/o?uploadType=multipart>: {
"error": {
"code": 403,
"message": "Insufficient Permission",
"errors": [
{
"message": "Insufficient Permission",
"domain": "global",
"reason": "insufficientPermissions"
}
]
}
}
: ('Request failed with status code', 403, 'Expected one of', <HTTPStatus.OK: 200>)

  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_plan.py", line 193, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 326, in core_dagster_event_sequence_for_step
    for evt in _type_check_and_store_output(step_context, user_event, input_lineage):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 380, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output, input_lineage):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 488, in _store_output
    handle_output_res = output_manager.handle_output(output_context, output.value)
  File "/usr/local/lib/python3.7/site-packages/dagster_gcp/gcs/io_manager.py", line 65, in handle_output
    retry_on=(TooManyRequests, Forbidden),
  File "/usr/local/lib/python3.7/site-packages/dagster/utils/backoff.py", line 64, in backoff
    raise to_raise
  File "/usr/local/lib/python3.7/site-packages/dagster/utils/backoff.py", line 58, in backoff
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/google/cloud/storage/blob.py", line 2858, in upload_from_string
    retry=retry,
  File "/usr/local/lib/python3.7/site-packages/google/cloud/storage/blob.py", line 2589, in upload_from_file
    _raise_from_invalid_response(exc)
  File "/usr/local/lib/python3.7/site-packages/google/cloud/storage/blob.py", line 4355, in _raise_from_invalid_response
    raise exceptions.from_http_status(response.status_code, message, response=response)

Following is the run configuration

execution:
  k8s:
    config:
      env_config_maps:
      - dagster-pipeline-env
      image_pull_policy: Always
resources:
  io_manager:
    config:
      gcs_bucket: my-bucket
      gcs_prefix: dagster
solids:
  multiply_the_word:
    config:
      factor: 2
    inputs:
      word: test

The pipeline can be executed successfully by simply retrying, but the same issue randomly reappear. Any idea what did go wrong?

Hey, I’m getting error

dagster not found

when I try to do multi-stage docker build using python3.7-slim as building image, and google gcr.io/distroless/python3:debug as running image

HI Dagster ninjas! I hate to ask such a simple question, but I didn't see an answer on SlackChannel or documentation.

Can I run two pipelines at the same time?

I have a pipeline that used to take 30 minutes that now takes 4 hours. Well, I need to run other pipelines I made because I depend on them, now. Can I just start another one, concurrently? Or do I have to wait for the other to finish? Full disclosure, I'm using DagIt to start the pipeline. Can I just start another at the same time, and track them in the "Runs" tab? I don't want to lose 2-3 hours due to a crash or something.

Hey folks! Is there a built-in way to clean up logs from old runs? We've been running a few experimental jobs in Dagster and are already up to 28GB in compute_logs (currently stored on NFS) and 1GB in event_logs (in Postgres). Keeping these around for a few days or weeks to debug failed pipelines makes sense, but left to grow forever could become prohibitively expensive. It seems like it would be pretty straightforward to supply our own

ComputeLogManager

and

EventLogStorage

implementations which would delegate to the native ones but also delete data for old runs periodically, but before I do this I want to see if a solution exists already.