dagster #dagster-support

Thomas Yopes

06/01/2021, 11:52 AM

Quick question: does anyone know how to define a

Nothing

dependency using

DependencyDefinition

that connects two composite solids? Use case I'm working with: Composite solid 1 runs N solids, composite solid 2 runs M solids, there's no obvious data dependency (it's lots of db / table setup work). I'm building the dependency dictionary object, but can't introduce an argument to the second composite solid that's

Nothing

(plus, even if I introduce an argument, I can't run the pipeline as this argument has to be used in the solids, which I would like to avoid) -- but in the documentation I seem to need an input value.

Raghu

06/01/2021, 2:46 PM

Hello Team! Am very much new here and trying to set up dagster as my go to tool for analytics workflows. ( relatively new to analytics as code, previously used GUI tools for ETL). It was easy for me to follow the cereals tutorials, to have a glimpse of how the tool works. Its very much fun to play around, until I want to set up a docker instance from scratch for my actual projects. My employer has compliance policies that doesn't allow me to import the entire https://github.com/dagster-io/dagster/ tree, but is there are guide to read the step by step process to set up dagster in docker/windows base machines besides this? The deployment section at https://github.com/dagster-io/dagster/tree/0.11.11/examples/deploy_docker was good, but am not sure how to set up only the files that are required to run the setup standalone on top of the python image provided by my engineering team. Hunting the slack channels here, online for blogs, videos but not luck. As am lacking software engineering background to set this up, any kind of guidance would greatly help me to get to the next level! Very much look forward to talk to someone who had such initial hiccups. I would definitely want to put up a how to deploy dagster using docker video post successful completion to help others.

Dan Corbiani

06/01/2021, 5:46 PM

@chris Is there a way to set a solid context parameter regarding what should be memoized? I've made some adjustments to the IO manager so only objects of a specific type are dumped to disk, but I'd like to make it so the user can specify which steps they would like to store. I tried to look in the context object within the IO manager but it didn't appear to carry along the solid configuration.

Jordan W

06/01/2021, 6:26 PM

Hello, I have been moving my data pipeline into the dagster workflow. Some of our solids are slow, so on top of refactoring the code, we want to use

async

where possible. I am having a little issue trying to get a basic working example to execute asynchronously. As of now, the basic example solids execute one after another. I saw in the docs that

execute_pipeline_iterator

might have to be used. I was wondering if there was any async execution for dagit/dagster cli (apart from

execute_pipeline_iterator

) and if I could get some help on my example. Appreciate any and all help.

Rei Tamai

06/02/2021, 8:21 AM

Hello Team! I’ve been playing around with Dagster lately. And I would like to load the data processed by Dagster into Snowflake as a table. I saw “https://docs.dagster.io/_apidocs/libraries/dagster-snowflake”. However, I am not quite sure how to use it. In particular, I don’t know how to set up the account information for Snowflake. Is there any more detailed information on how to use this library? (I`m Japanese, so I think the on-time is quite different.)

sk4la

06/02/2021, 12:23 PM

Hiii, does anyone know how to serve more than one Python package using the gRPC server (i.e.

dagster api grpc

) ? I currently load my repositories locally from multiple packages using the

- python_package: ...

stanza in

workspace.yaml

; but it does not seem possible using the gRPC server, as it does not make use of the local

workspace.yaml

file. Any ideas?

Thomas Yopes

06/02/2021, 12:24 PM

Hello! Has anyone setup up a working version of a scheduler that can build/edit the pipeline before executing it? I have written the functionality required to produce the entire pipeline (solids, solid configs, dependencies), but I'd like to be able to adjust it when the scheduler runs. Let me know 👍

Dan Corbiani

06/02/2021, 12:51 PM

@chris We've been continuing our memoization work. All is mostly good except for the DagsterNoStepsToExecute error. Is it intentional that an exception is raised if there is nothing to do? Am I calling the pipeline incorrectly? I end up having to wrap a bunch of our execution processes in try blocks like below: It makes me wonder how to access the output from a previously run memoized pipeline.

Andrew Herbst

06/02/2021, 3:36 PM

Hi all, is it possible to have order-based (nothing) dependencies that have can consume a real parameter as well? i.e.,

solid1 -> Nothing, solid2 -> String, solid3(solid1(), solid2())

When I try this I get an error about unmapped parameters, I have a gist that illustrates this issue if that would help.

Dylan Bienstock

06/02/2021, 5:03 PM

Hi! I have been using the DagsterGraphQLClient to submit pipeline runs and get a pipelines run status. I create the client like so

client = DagsterGraphQLClient("localhost", port_number=3000)

and it is working when I run dagit locally. However, when i deploy it to my dagster k8 instance it is not able to connect. Is there a different port I should be using? Or do I need to open something up to allow it to communicate?

John Conwell

06/02/2021, 5:22 PM

One thing I’ve been struggling with in Dagster is how to build dynamic DAGs based on some configuration state; like values in a resource config. This is quite common in ML experiment workflows where based on the config values the set of solids I want to execute are different. Take this pipeline for example:

Copy code

@pipeline(mode_defs=[cluster_mode_def])
def build_models():
    # determine if we're building or copying the training data set from production
    if config['BUILD_DATA_SET']:
        training_data_set = cmds.build_training_data_set(cmds.training_data_config())
    else:
        training_data_set = cmds.copy_training_data_set()

    # build vectors for each feature family, build full vector space, then build the model
    cmds.build_model(
        cmds.build_feature_space(
            cmds.build_feature_vectors(
                training_data_set
            )
        )
    )

Based on some config state I’d want to either build my training dataset or just copy it from a production environment. But since pipelines don’t execute, just build the solid DAG structure and don’t have access to any configuration information, how would I do this?

Eduardo Santizo

06/02/2021, 6:12 PM

Hey! I've been using the "dagster docker compose" example and it's working well. Now I want to change code inside the "pipelines" container without updating the image and rebuilding the container. I managed to make changes by entering the container and modifying the files, but Dagit doesn't seem to pickup the changes. Any solution for this?

Tom Han

06/02/2021, 6:30 PM

Hi dagster team, I have an error that I cannot figure it out, please help. The problem occurs when the @daily_schedule try to run at execution time each day. The strange thing is when I manually run the backfill partition from this @daily_schedule, it ran without any problems. The error shows the following:

Copy code

dagster.check.CheckError: Invariant failed. Description: Attempted to deserialize class "ScheduleExecutionData" which is not in the whitelist.
  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 57, in sync_get_external_schedule_execution_data_grpc
    else None,
  File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 278, in external_schedule_execution
    "".join([chunk.serialized_chunk for chunk in chunks])
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 242, in deserialize_json_to_dagster_namedtuple
    check.str_param(json_str, "json_str"), whitelist_map=_WHITELIST_MAP
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 252, in _deserialize_json_to_dagster_namedtuple
    return _unpack_value(seven.json.loads(json_str), whitelist_map=whitelist_map)
  File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 278, in _unpack_value
    f'Attempted to deserialize class "{klass_name}" which is not in the whitelist.',
  File "/usr/local/lib/python3.7/site-packages/dagster/check/__init__.py", line 167, in invariant
    raise CheckError(f"Invariant failed. Description: {desc}")

My running environment is standard k8s on the cloud, and the schedule code is following:

Copy code

@daily_schedule(
    pipeline_name='pipeline_daily_partition',
    execution_timezone="Asia/Shanghai",
    execution_time=datetime.time(hour=23, minute=41),
    start_date=datetime.datetime(2010, 5, 27),
    partition_days_offset=0,
    mode=prd_partition_preset.mode,
)
def daily_schedule_partition(date):
    def nested_update(d: dict, target, value):
        for k, v in d.copy().items():
            if isinstance(v, dict):
                nested_update(v, target, value)
            else:
                if d.get(target):
                    d[target] = value
    modified_run_config = prd_partition_preset.run_config.copy()
    nested_update(modified_run_config, 'date', date)
    return modified_run_config

Jordan W

06/02/2021, 7:47 PM

Hello, quick question, I was wondering if it is possible to use

dagster.mem_io_manager

when running asynchronous solids? I seem to be getting an error when using this as my io manager. When I use the other default

fs_io_manager

it does work. Thanks in advanced!

Marco

06/03/2021, 8:32 AM

Hello, is there a way to access the partition within a failure_hook’s context? I’d like to send a message for ‘live’ runs, where live is defined by partition = today’s date

Thomas Yopes

06/03/2021, 11:53 AM

Does anyone know how to access the metadata in an asset materialization when I grab events with

events_for_asset_key

? Can't seem to find it in the examples anywhere. I'm working with the example code in the sensor as follows:

record_id, event = events[0]

But I'm wondering if there's a way to get the metadata I attached to the asset.

wangm23456

06/03/2021, 1:21 PM

I made changes to the example in the documentation(https://docs.dagster.io/concepts/solids-pipelines/pipelines), and I found that if I used a branch solid, then the unexecuted solid would block the other solids below it, even though their dependencies were all satisfied.

wangm23456

06/03/2021, 1:21 PM

Copy code

import random

from dagster import Output, OutputDefinition, pipeline, solid, composite_solid


@solid(
    output_defs=[
        OutputDefinition(name="branch_1", is_required=False),
        OutputDefinition(name="branch_2", is_required=False),
    ]
)
def branching_solid_test(context):
    num = 0
    <http://context.log.info|context.log.info>(str(num))
    if num == 0:
        yield Output(1, "branch_1")
    else:
        yield Output(2, "branch_2")


@solid
def branch_1_solid(_input):
    return 1


@solid
def branch_2_solid(_input):
    return 2

@solid
def s1_solid(_input):
    return 1

@solid
def s2_solid(_input):
    return 2

@composite_solid
def solid1(_input):
    s = branch_1_solid(_input)
    s = s1_solid(_input)
    return s

@composite_solid
def solid2(_input):
    s = branch_2_solid(_input)
    s = s2_solid(_input)
    return s

@solid
def final(_s):
    return _s

@pipeline
def branching_pipeline():
    branch_1, branch_2 = branching_solid_test()
    s = solid1(branch_1)
    s = solid2(branch_2)

    final(s)

wangm23456

06/03/2021, 1:23 PM

Run the above code and final solid will never be executed.

wangm23456

06/03/2021, 1:26 PM

The documented example only implements branches, which do not match the schematic.

Akshay Verma

06/03/2021, 3:04 PM

Hi all, We are looking into using Dagster for our workflow management. Of the several requirements, the following are the most important is: • to launch pipelines from Frontend -> I think this can be done from GraphQl API • to create custom DAGs from the frontend -> this can be achieved if the frontend sends YAML or JSON config or through GraphQl API • To be able to orchestrate a DAG from the Solids spread across different containers (which have different programming languages like Python, Java, Julia and interact using REST APIs) If you guys could please point me to docs/resources which could help us in solving these points with Dagster, then it would be helpful to create a strong use-case. (Let me know if this is not the correct channel)

Ethan Brown

06/03/2021, 4:01 PM

Hello! I'm curious about code versioning for solids and pipelines. If I have a pipeline that runs successfully at one time, can I freeze that version to something like a git commit? Essentially I am thinking about having the ability to have a stable version in production while continuing to develop and push changes. What are best practices for code versioning where you have multiple stable production pipelines?

Brian Seo

06/03/2021, 5:42 PM

Is there a way to make a date variable created by a schedule into solid configs available to all solids in the pipeline instead of manually specifying which solids to pass the date to? I can see passing it to just a single solid in the first step and sending it to all downstream solids in a pipeline but that design pattern feels extremely clunky.

Matyas Tamas

06/03/2021, 8:00 PM

Low-priority, but has anyone else found the dagit partition backfill interface to be extremely slow when you have a large-ish number of partition? I'm working with maybe 2-3k partitions on a dataset and it takes several minutes for the UI to show the status of all the partition runs (I'm trying to figure out what partitions need to be re-run). I've just tried this on our dev environment which uses sqlite and local storage for backends.

Arun Kumar

06/03/2021, 8:26 PM

Hi team, is it possible to set different tag values for different runs of the same pipeline? If yes, is there a way to search and filter the runs in the Dagit by the tag values?

Andrew Herbst

06/03/2021, 9:31 PM

Hi all, I’m bumping into a GraphQL error when attempting to fetch the logs for a run, this is against v0.11.10. I have dropped the associated stack trace in this gist https://gist.github.com/aherbst-broad/b44183704ef8dfaf0b4336a87e8b4d8a

David

06/04/2021, 7:04 AM

Hello! If a solid is retried by

raise RetryRequested(max_retries=2)

, is there a way to determine within the solid that it is running as a retry, and how many times it has already already been retried? Thanks!

Thomas Yopes

06/04/2021, 9:13 AM

Is there any chance there's future support for visualizing an executed pipeline in dagit that was called with

execute_pipeline

from a solid in another pipeline? I've tried lots of different ways to get Dagster to run a schedule for a pipeline that needs to be re-generated fresh when the scheduler starts, but haven't been able to make any progress (lots of config, timing, and invariant errors -- tried sensors with no luck, assets, etc). For the time being, I'm simply going to have a helper pipeline that calls the code required to generate the dynamic pipeline then calls

execute_pipeline

and it would be great if this dynamic pipeline appeared in the runs tab. Let me know! or let me know if you're someone who's managed to get Docker + Dagster working with dynamic pipeline generation (note: I don't mean the dynamic output but defining the entire pipeline up front before it's executed).

Denis Rakhimov

06/04/2021, 10:32 AM

Queue problem Very few tasks in progress in the queue Runs only 2-5 tasks

Darren Haken

06/04/2021, 2:57 PM

Hey folks, I’m new to Dagster and I’ve been going through the tutorials. My org is using Airflow at the moment and we’re exploring Dagster to understand the benefits it could offer. I am trying to create a pipeline using the Data Bricks package and solid. Is someone able to offer any help?