dagpurr  Hi all! Dagster community meeting is starting in 20 minutes at 9 am Pacific Time (UTC-7). Link to the event We have two presentations scheduled for the meeting about their production Dagster setups: • @dhume from Drizly will demonstrate how Drizly configures its Dagster deployments to support a heterogeneous team • @Noah K from Geomagical Labs will discuss Geomagical’s approach to building customer-facing applications. There will also be a Q&A and discussion section for any questions you have. If you have never registered our events before, please DM me your email or use Eventbrite: Link to the event page to register for the event. If you aren’t able to make it, we’ll share a recording on Youtube afterward.

dagpurr Hi all! After the meeting we would like to hear your feedback. Feel free to submit your feedback here. Thanks our speakers @dhume @Noah K for the great presentations today!

My slides are up at https://speakerdeck.com/coderanger/dagster-and-geomagical if they are useful to anyone.

Hi all! What is the difference between using

configured

to curry some configuration about a solid vs using a

composite_solid

? Is there a best practice when you want to curry inputs or partially configure a solid?

A composite solid can have several solids inside it

While configured() is just setting stuff on one

composites are more like mini sub-pipelines

That’s what I also thought, but the configured object doesn’t really behave like a solid afterward. I posted a bug report, but then I started to wonder if my mental model is wrong. Eg https://github.com/dagster-io/dagster/issues/3662

Hey 👋 Can we pass a

DynamicOutput

to a

composite_solid

? My pipeline works fine when I pass the

DynamicOutput

to a single

solid

, but I would like the

.map()

to execute more than one solid, with the general flow being; 1. Fetch a list of IDs from the database (unknown before execution time); 2. For each; query an API, build a dataframe, output to database But I’m getting the following error when I package step 2. as a `composite_solid`;

dagster.core.errors.DagsterSubprocessError: dagster.check.CheckError: Member of list mismatches type. Expected <class 'dagster.core.execution.plan.inputs.StepInput'>. Got UnresolvedStepInput(name='id', dagster_type_key='Int', source=FromPendingDynamicStepOutput(step_output_handle=StepOutputHandle(step_key='query_records', output_name='id', mapping_key=None), solid_handle=SolidHandle(name='do_multiple_steps', parent=None), input_name='id')) of type <class 'dagster.core.execution.plan.inputs.UnresolvedStepInput'>.

Hi, I have dagster deployed and running on version

0.10.4

. Everything runs fine, except the scheduler seems to continually shut down after about 2-3 hours with the following error (pasted below). It seems that I have to restart the daemon continually to address this. is this normal? is there a way to suppress these errors? I'm invoking

daagster-daemonn

via

supervisord

with the simple

run

commannd.

dagster.serdes.ipc.DagsterIPCProtocolError: Timeout: read stream has not received any data in 15 seconds
  File "/usr/local/lib/python3.8/site-packages/dagster/scheduler/scheduler.py", line 86, in launch_scheduled_runs
    with RepositoryLocationHandle.create_from_repository_location_origin(
  File "/usr/local/lib/python3.8/site-packages/dagster/core/host_representation/handle.py", line 57, in create_from_repository_location_origin
    return ManagedGrpcPythonEnvRepositoryLocationHandle(repo_location_origin)
  File "/usr/local/lib/python3.8/site-packages/dagster/core/host_representation/handle.py", line 192, in __init__
    self.grpc_server_process = GrpcServerProcess(
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 1037, in __init__
    self.server_process = open_server_process(
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 942, in open_server_process
    wait_for_grpc_server(server_process, output_file)
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 878, in wait_for_grpc_server
    event = read_unary_response(ipc_output_file, timeout=timeout, ipc_process=server_process)
  File "/usr/local/lib/python3.8/site-packages/dagster/serdes/ipc.py", line 39, in read_unary_response
    messages = list(ipc_read_event_stream(output_file, timeout=timeout, ipc_process=ipc_process))
  File "/usr/local/lib/python3.8/site-packages/dagster/serdes/ipc.py", line 152, in ipc_read_event_stream
    raise DagsterIPCProtocolError(

How come

postgres_url

in

pg_config

is a str instead of a StringSource? This would allow setting the entire url via environment variable in the yaml, the postgres_db is setup like this:

"postgres_db": {
                "username": StringSource,
                "password": StringSource,
                "hostname": StringSource,
                "db_name": StringSource,
                "port": Field(IntSource, is_required=False, default_value=5432),
            },

It looks like this would work, but it's static?

run_storage:
  module: dagster_postgres.run_storage
  class: PostgresRunStorage
  config:
    postgres_url: "<postgresql://test:test@{hostname}:5432/test>"

Hi All ! I am scheduling a pipeline in dagster to run at a specific time daily using

@daily_schedule

however when its time comes to run, it does not run instead scheduler simply skips this pipeline, however i run the same pipeline using

cron_schedule

and scheduler runs this pipeline in this case !!! 🤐 Here is the screenshot: Thank You!

Run solids / workflow /whatever in a controlled python environment Hello, I still try to figure out all pros and cons using dagster. I have this question. Is dagster in production able to "conternize" easily at one of those levels: • solid • workflow • other ? My use case is that I want to run some ML models (any frameworks) into dagster as solid. Since dependencies can be difficult if you have to install multiple environment, what is the capacity of dagster for this ? I can find workaround but if it is already there... 🙂

Hi Everyone! I am getting strange error while setting up concurrency configurations: dagster.yaml

scheduler:
  module: dagster.core.scheduler
  class: DagsterDaemonScheduler


run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 25
    tag_concurrency_limits:
      [
        { key:"test", value:"two", limit:2 }
      ]

While running the program it says following:

raise DagsterInvalidConfigError(
dagster.core.errors.DagsterInvalidConfigError: Errors whilst loading configuration for {'max_concurrent_runs': Field(<dagster.config.config_type.Int object at 0x7fd295c8ea30>, default=@, is_required=False), 'tag_concurrency_limits': Field(<dagster.config.config_type.Noneable object at 0x7fd29b096dc0>, default=@, is_required=False), 'dequeue_interval_seconds': Field(<dagster.config.config_type.Int object at 0x7fd295c8ea30>, default=@, is_required=False)}.
    Error 1: Received unexpected config entries "['key:"test"', 'limit:2']" at path root:tag_concurrency_limits[0]. Expected: "['key', 'limit', 'value']."
    Error 2: Missing required config entries "['key', 'limit']" at path root:tag_concurrency_limits[0]".
    Error 3: Received unexpected config entries "['key:"test1"', 'limit:5']" at path root:tag_concurrency_limits[1]. Expected: "['key', 'limit', 'value']."
    Error 4: Missing required config entries "['key', 'limit']" at path root:tag_concurrency_limits[1]".

I am using the exact syntax based on docs: https://docs.dagster.io/overview/pipeline-runs/limiting-run-concurrency#main

Hi, I'm currently in the process of deploying Dagster locally. Everything was working smoothly until I decided to change the file structure a bit. I am still able to run dagit but the daemon service now throws an error when looking for schedules. Here's the error: FileNotFoundError: [WinError 3] The system cannot find the path specified. Seems like the daemon service is looking into the old structure even though I edited the workspace.yaml. How do I fix this?

Best Practice Question regarding type-hints with DataFrames: Can you derive a

DagsterDataType

from a

PysparkDataFrame

. I have a generic solid

load_delta_table_to_df

, but in my Pipeline I'd like to type-check that the returned DataFrame has certain columns (not always the same see example attached). I try to achieve that with custom DagsterType

NpsDataFrame

and

TagDataFrame

in my pipeline (see attachment), but that will not show the type in Dagit. How could I use a generic solid but returning different typed DataFrames? I'd like to see NpsDataFrame and TagDataFrame instead of generic PySparkDataFrame. Any best practices? Or should I add an additional parameter to

load_delta_table_to_df

where I define the output DataFrame? Thanks a lot guys!

Whats everyone using for code-review tools? We're using upsource but I'm evaluating other options and was curious to see what other folks prefer.

Hi, is there anyway to launch the grpc server with some sort of autoreload? I'm running it from a docker container with a mounted volume with my source code but I need to restart the container continually so it picks up the changes...

In migration from 0.9.3 to 0.10.4 we're seeing a typing error:

dagster.core.errors.DagsterInvalidDefinitionError: Invalid type: dagster_type must be DagsterType, a python scalar, or a python type that has been marked usable as a dagster type via @usable_dagster_type or make_python_type_usable_as_dagster_type: got typing.NoReturn

on something that was previously working. Is there a dagster typed equivalent of NoReturn or should I mark NoReturn as usable?

Hey all, here’s the recording of yesterday’s community meeting:  All the slide decks from the presentations are linked in the description. If you have never registered before and would like to join our monthly community meeting, sign up for an invite here: http://bit.ly/march-cm

Hi all! Is there a way to hook into the dagster config type validation mechanism? I’ve been using pydantic for validating types, and that works when combining dagster type loaders + the config “inputs” section. However validation errors are discovered only at runtime. Is there a way to define custom field validators through eg a function? Alternatively, to define custom dagster config types?

In 0.9.3-0.10.4 migration I'm also seeing the below error, which is quite vexing as I don't think it indicates where the problem is being generated:

Operation name: RunsRootQuery

Message: Exactly 5 or 6 columns has to be specified for iteratorexpression.

Path: ["repositoriesOrError","nodes",1,"schedules",1,"futureTicks"]

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

I'm hopeful that moving off the system cron scheduler will remediate, but any guidance is nonetheless helpful

Would a deploy guide for Heroku be too vendor-specific and better suited to a blog post? I've got our dagster stack running on Heroku, it was pretty straight forward and would be happy to write up how to do it.

Can you set env variables in

workspace.yaml

? I tried similar syntax to the

dagster.yaml

and that didn’t seem to work

Hi, I am new to Dagster. Is there any way to configure a pipeline so that a certain solid within it is executed/not executed based on a boolean?

is there a recommended process for schedule changes (or maybe this is a user error but a cryptic error message)? i changed the schdedule intervals for my pipelines and im getting this GraphQL error:

Operation name: SchedulesRootQuery

Message: Invariant failed.

Path: ["repositoryOrError","schedules",0,"futureTicks"]

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

Stack Trace:
  File "/usr/local/lib/python3.8/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.8/site-packages/graphql/execution/executors/sync.py", line 16, in execute
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/dagster_graphql/schema/schedules/schedules.py", line 96, in resolve_futureTicks
    tick_times.append(next(time_iter).timestamp())
  File "/usr/local/lib/python3.8/site-packages/dagster/utils/schedules.py", line 27, in schedule_execution_time_iterator
    check.invariant(len(cron_parts) == 5)
  File "/usr/local/lib/python3.8/site-packages/dagster/check/__init__.py", line 172, in invariant
    raise_with_traceback(CheckError("Invariant failed."))
  File "/usr/local/lib/python3.8/site-packages/future/utils/__init__.py", line 446, in raise_with_traceback
    raise exc.with_traceback(traceback)

Newbie question on types: why not use the PEP484 syntax? (Quote from the docs: https://docs.dagster.io/overview/types)

The Dagster type system is independent from the PEP 484 Python type system, although we overload the type annotation syntax on functions to make it easier to specify the input and output types of your solids.

Hello, I'm trying to run my Dagster setup inside Docker. I'm following this example: https://docs.dagster.io/examples/deploy_docker I downloaded this example and ran

docker-compose up

but something is not working. I'm not getting any errors however the application is not available at localhost:3000. Just to confirm, should this example be runnable as-is or is any other configuration required? Given that I'm not getting any errors, I'm thinking the problem may be with exposing port 3000. I also ran

lsof -i:3000

locally and it shows nothing, making me think the container port isn't being exposed properly. The docker compose file does include the line

expose: - "3000"

in the dagit container though, so not sure what is wrong. Anybody have any suggestions? Thanks so much!

Chris DeCarolis just published a new version: 0.10.5. See https://github.com/dagster-io/dagster/releases/tag/0.10.5 for the release notes.