An additional (possibly dumb) question - does Dagit/dagster support mode alteration un the UI/CLI? For instance if I wanted to run a pipeline under a specific mode, but with one resource changed? Or do I have to create a new mode to support that use-case?

Hi, I have probably a very silly question related to sensors. I can run my pipeline with a preset config using this command dagster pipeline execute -f pipeline_script.py --preset default I have a sensor (a copy of my_s3_sensor from dagster examples) to monitor files in the bucket. How do I set parameters for my run_config in the sensor to call this pipeline ( say example_pipeline). In the dagster examples, the run_config parameter includes solids. I am a bit confused here as I only need to run my pipeline , since all the solid dependencies are defined there. Also, it says Expected: "['execution', 'intermediate_storage', 'loggers', 'resources', 'solids', 'storage']." for config entries. Thanks in advance!

set the channel topic: Next community meeting May 11 2021! Sign up here: https://forms.gle/Znt4fBt8tpf8tVnN9. Agenda here: https://bit.ly/2Pqz3J8. Subscribe to our Youtube channel http://bit.ly/2XMyY3z for past events videos

Isaac Hellendag just published a new version: 0.11.3. See https://github.com/dagster-io/dagster/releases/tag/0.11.3 for the release notes.

Hi! Can you, please, advise me, how to run solid by name in some conditions… For more clear explanation, I will try to explain by some pseudo-code.

@solid
def s1(context):
               pass
@solid
def s2(context):
               pass
 
If _var==1:
    _run_solid_by_string_name('s1')
else:     
	_run_solid_by_string_name('s2')

Thank you

Hi all! We are looking to build a pipeline that has two branches that in turn fanout using the new dynamic mapping & collect feature. We’d like to collect() on both branches and do some post-processing. However, when we attempt to collect on the two branches and feed those as input into a downstream solid we bump into this error:

collect index unexpectedly set twice

. So, we’ve introduced an intermediate “fan-in” solid on both branches that collects the results and then hands those off to the downstream post-processing solid. My question: is this a known issue and is the notion of introducing the intermediate fan-in solid the correct approach here?

Hi Everyone, Can someone suggest great courses for Dagster?

I have a very simple pipeline, adapted from the

deploy_docker

example that fails (see: https://github.com/jeremyadamsfisher/dagster-sensor-min-failing)

from dagster import pipeline, repository, schedule, solid, sensor, SkipReason


@solid
def hello(_):
    return 1


@pipeline
def my_pipeline():
    hello()


@schedule(cron_schedule="* * * * *", pipeline_name="my_pipeline", execution_timezone="US/Central")
def my_schedule(_context):
    return {}


@sensor(pipeline_name="my_pipeline")
def always_skips(_context):
    yield SkipReason("I always skip!")


@repository
def deploy_docker_repository():
    return [my_pipeline, my_schedule, always_skips]

This is the traceback:

docker_example_daemon        | 2021-04-03 19:36:56 - SensorDaemon - INFO - Checking for new runs for sensor: always_skips
docker_example_daemon        | 2021-04-03 19:36:56 - SensorDaemon - ERROR - Error launching sensor run: json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
docker_example_daemon        | 
docker_example_daemon        | Stack Trace:
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/daemon/sensor.py", line 230, in execute_sensor_iteration
docker_example_daemon        |     sensor_debug_crash_flags,
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/daemon/sensor.py", line 264, in _evaluate_sensor
docker_example_daemon        |     job_state.job_specific_data.last_run_key if job_state.job_specific_data else None,
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/core/host_representation/repository_location.py", line 448, in get_external_sensor_execution_data
docker_example_daemon        |     last_run_key,
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/api/snapshot_sensor.py", line 41, in sync_get_external_sensor_execution_data_grpc
docker_example_daemon        |     last_run_key=last_run_key,
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/grpc/client.py", line 291, in external_sensor_execution
docker_example_daemon        |     res.serialized_external_sensor_execution_data_or_external_sensor_execution_error
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 236, in deserialize_json_to_dagster_namedtuple
docker_example_daemon        |     check.str_param(json_str, "json_str"), whitelist_map=_WHITELIST_MAP
docker_example_daemon        |   File "/usr/local/lib/python3.7/site-packages/dagster/serdes/serdes.py", line 246, in _deserialize_json_to_dagster_namedtuple
docker_example_daemon        |     return _unpack_value(seven.json.loads(json_str), whitelist_map=whitelist_map)
docker_example_daemon        |   File "/usr/local/lib/python3.7/json/__init__.py", line 361, in loads
docker_example_daemon        |     return cls(**kw).decode(s)
docker_example_daemon        |   File "/usr/local/lib/python3.7/json/decoder.py", line 337, in decode
docker_example_daemon        |     obj, end = self.raw_decode(s, idx=_w(s, 0).end())
docker_example_daemon        |   File "/usr/local/lib/python3.7/json/decoder.py", line 355, in raw_decode
docker_example_daemon        |     raise JSONDecodeError("Expecting value", s, err.value) from None

Am I missing something or should I file a github issue?

Has any work been done to date to support executions via knative (and therefore google cloud functions)?

Hi everyone, I am new in Dagster. Any buddy can tell me, what's the best capability in Dagster ?

Hey. I was trying to implement the "error detecting sensor" in the dagster docs, but I'm getting a "PipelineRunsFilter" error during its execution. Any idea why this is happening?

hello, I'm wondering if anyone has tried to pickle a solid before or if maybe there's a better way to serialize these for context, we are trying to pickle

{Solid,CompositeSolid,Pipeline}ExecutionResult

objects and solids are contained as attributes in these simple example:

import pickle
import dagster


@dagster.solid
def test(context):
    pass

with open('/home/chet/data/testpickle.pkl', 'w') as p:
    pickle.dump(test, p)

---------------------------------------------------------------------------
PicklingError                             Traceback (most recent call last)
<ipython-input-45-e900432760d1> in <module>
     13 
     14 with open('/home/chet/data/testpickle.pkl', 'w') as p:
---> 15     pickle.dump(test, p)
     16 
     17 

PicklingError: Can't pickle <function test at 0x7f9bf70ab9d8>: it's not the same object as __main__.test

I am really excited to dig into the new

collect

and dynamic mapping feature. I wanted to share my use case and make sure it fits. I have a script that passes in a list of IDs, and iterates over each ID, to hit an api and download a file for that ID. The list of IDs varies per run. I also use

concurrent.futures

to use multiple threads while downloading. From looking at the docs and using the test code, I really like how it fans out each interaction to see the dynamic nature of what work the solid is doing. Is collect/dynamic mapping meant to run in serial? I also saw a mention of “retries” in the 0.11.0 blog post - is that an automatic thing? I’d love any feedback on my use case - whether y’all think it fits and/or any gotchas I might consider. Thx! (Please pardon any incorrect use of technical terminology 🙂)

Hi all, Happy Easter. Is ECS on AWS a viable deployment option for Dagster yet? I heard some things are still getting setup that means it's not as well supported Kubernetes? Is this right, if so does anyone have details? Thanks

Basically the ECSRunLauncher - does it mean the dagster-daemon doesn't work?

Hi all 👋 Is there a standard pattern for initializing singleton-like resources? I'd like to be able to write a resource definition that initializes a given resource only once and reuses that initialization for the resources provided to solids in other executors. This is useful when initializing a resources is side-effecting: e.g., you create an external resource, and are given back a UUID for that resource, and then want that UUID available to use that external resource in other parts of the pipeline. (This might be a bit abstract, but I'll add an example in the comments).

Hi How python dependencies (normally in requirements.txt) have to be added to user scripts in helm (k8sScheduler/celeryK8sRunLauncher) deployment? Do we have to add requirements manually/ it is handled automatically (like using requirements.txt etc) Maybe somebody can give hint where to look/ how to solve this. For example, in user scripts I am trying to use newsfetch library but having error:

Hello guys, I'm using Dagster 0.9.22 (and running it inside a container) and looking to reduce the size of database because the "events_logs" table is now over 60GB. I have some questions: • does the CLI "dagster wipe" delete the entire data or can I specify a date range? • which tables should I drop? If I only delete data from "events_logs" will my service still work? Thank you in advance and have a nice day!

Hi, I've been installing dagster via poetry and specified certain packages with an older version (e.g.

dagster-pyspark = "==0.10.9"

and

dagster-celery-k8s = "==0.10.9"

). However, when the dependencies are resolved, the 0.10.9 versions of the specified packages are installed but the newest versions of dependencies such as dagster-spark, dagster-k8s and dagster-celery (currently 0.11.3) which were not explicitely specified. As I understand all versions of dagster packages should always be identical. Of course I can add dagster-spark, dagster-k8s and dagster-celery to my requirements file (pyproject.toml in poetry) but I can't foresee future changes or additions in dependencies for dagster packages, so I'd have to do that every time I add a package or update a version. Is it possible that the dagster team makes sure that these mismatches don't occur by preventing dependency resolution if versions of two or more dagster packages are unidentical/incompatible, and by locking the versions of dependencies to the specified version of the corresponding dagster package (i.e. if I specify

dagster-pyspark = "==0.10.9"

,

dagster-spark = "==0.10.9"

is implied? Or am I missing something?

Hi, recently I've been experimenting with the dagster-daemon and an error related to the scheduler comes up now and then that basically resolves itself after some time and a refresh. Is this a bug or am I missing something?

Hi all, If I choose a remote PostgresSQL database for Dagster's logs etc, that's accessed over the public internet (i.e. not sitting in a VPC), does anyone know if Dagster automatically encrypts the data with TLS during transit? I assume it does, but I thought I would check Thanks

Hi Is there any way to combine preset and run config? I want to build pex zip of my pipeline with preset config yamls inside. And then do spark submit of this zip, but would like to override some values defined in that yamls. If i use pipeline execute cli i am getting error that preset and config cannot be specified at the same time. Thank you.

with dynamic pipelines, I'm trying to get a dynamic schedule to work, and it renders okay. but i'm trying to render a dynamic pipeline as follows:

@schedule(
                    pipeline_name=yaml_file["name"],
                    name=yaml_file["name"] + "_schedule",
                    cron_schedule="* * * * *",
                    mode="prod",
                    execution_timezone="UTC",
                )
                def _foobar_schedule(date):
                    name = eval(yaml_file["name"])
                    name.get_preset("prod").run_config

but I get an error that

NameError: name 'mypipelinename' is not defined

Is it possible to call a dynamic pipeline to execute?

should i use

dagster.execute_pipeline

?

version 0.11.3: So I have a pipeline that is normally ran on a schedule for 8:05 am, but didn't actually run due to network drop. So after daemon was re-launched, the pipeline didn't actually run until 11:11am. Not sure what the tick history is suppose to show, but it is showing that pipeline ran at 8:05am. Is that expected or normal?

Hey, I am upgrading to v0.11.3 and now when using docker every module import fails. This seems to be related to the working directory I am using, but it has been working ever since 0.10, did anything change in latest release?

I am using WORKDIR /opt/dagster/app

I keep getting the following error when navigating to the partition tab of one of my pipelines:

Operation name: PartitionSetLoaderQuery

Message: Unknown fragment "PartitionGraphFragment".

Path: 

Locations: [{"line":69,"column":6}]

I am able to run the backfill with no problems but my screen is full of GraphQL error notifications, anyone knows why does this happen?

Hi, for Dagster's Postgres storage, if using an existing Postgres instance should we setup a brand new database, and then will Dagster populate it with the relevant tables provided the relevant database details are in the YAML file? If we use an existing database, rather than a fresh one, will it automatically create it's own schema or what exactly will happen? Thanks