dagster #dagster-support

Daniel Suissa

01/04/2022, 8:01 PM

Hi, resurfacing a question I raised during the holidays: Is it possible to debug dagster in pycharm? I'm calling

pydevd_pycharm.settrace

at the top of my @job definition and using

in_process_executor

but my job seems to hang when I do that. The debugger does connect though. Wondering if the Dagster contributors know anything about this or if anyone else in the community managed to do this.

Ryan Riopelle

01/04/2022, 9:05 PM

Seeing an issue with runner pods when trying to run a custom image, kubectl describe pods shows this as the error. "/usr/local/bin/python3: /usr/local/bin/python3: cannot execute binary file, Exit Code: 126" Has anyone run in to this issue and know how to solve? I am using Bazel as build tool which might create the issue, but all the files are 777.

Mark Kudryk

01/04/2022, 11:28 PM

I’m using version 0.12.12. I’m using sensors to detect when one pipeline is completed to kick off another pipeline. In one of my sensors I’m returning a SkipReason with a message indicating why. In Dagit in the sensor the message that I’m expecting is not being shown in the tick pop-up; the message is getting output to the console, but I’d hope that it would show up in Dagit. Should it not be? I’ve looked to see if there are any issues on this but could not find anything.

Ryan Riopelle

01/05/2022, 12:28 AM

Seeing this error when I try to run from dagster_cli. Any thoughts? Command is "`dagster job execute --working-directory /path_to_dir -f path/to/repo.py --job hello_cereal_shield_job`" on an instance separate from daemon or dagit.

Ryan Riopelle

01/05/2022, 12:28 AM

Copy code

2022-01-04 23:45:38 - dagster - ERROR - hello_cereal_shield_job - b1817925-525e-4768-ae49-2937cf528ab7 - 610 - RUN_FAILURE - Execution of run for "hello_cereal_shield_job" failed. An exception was thrown during execution.

RuntimeError: 
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':
                freeze_support()
                ...

        The "freeze_support()" line can be omitted if the program
        is not going to be frozen to produce an executable.

Stack Trace:
  File "/app/web/piper/hmc/dagster_cli_bin.runfiles/pypi__36__dagster_0_13_4_linux_x86_64/dagster/core/execution/api.py", line 774, in pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/app/web/piper/hmc/dagster_cli_bin.runfiles/pypi__36__dagster_0_13_4_linux_x86_64/dagster/core/executor/multiprocess.py", line 163, in execute
    event_or_none = next(step_iter)
  File "/app/web/piper/hmc/dagster_cli_bin.runfiles/pypi__36__dagster_0_13_4_linux_x86_64/dagster/core/executor/multiprocess.py", line 268, in execute_step_out_of_process
    for ret in execute_child_process_command(command):
  File "/app/web/piper/hmc/dagster_cli_bin.runfiles/pypi__36__dagster_0_13_4_linux_x86_64/dagster/core/executor/child_process_executor.py", line 140, in execute_child_process_command
    process.start()
  File "/usr/local/lib/python3.6/multiprocessing/process.py", line 105, in start
    self._popen = self._Popen(self)
  File "/usr/local/lib/python3.6/multiprocessing/context.py", line 284, in _Popen
    return Popen(process_obj)
  File "/usr/local/lib/python3.6/multiprocessing/popen_spawn_posix.py", line 32, in __init__
    super().__init__(process_obj)
  File "/usr/local/lib/python3.6/multiprocessing/popen_fork.py", line 19, in __init__
    self._launch(process_obj)
  File "/usr/local/lib/python3.6/multiprocessing/popen_spawn_posix.py", line 42, in _launch
    prep_data = spawn.get_preparation_data(process_obj._name)
  File "/usr/local/lib/python3.6/multiprocessing/spawn.py", line 143, in get_preparation_data
    _check_not_importing_main()
  File "/usr/local/lib/python3.6/multiprocessing/spawn.py", line 136, in _check_not_importing_main
    is not going to be frozen to produce an executable.''')

01/05/2022, 12:52 AM

Dagster version: 0.13.12 A couple of problems: 1. I am unable to specify max_concurrent, even though multiprocess executor is default for this version of Dagster. 2. My runs do not appear to be in multiprocess mode, the ops run in serial.

Nilesh Pandey

01/05/2022, 11:59 AM

Hi, Let's say I have two solids in a pipeline. They are independent of each other. Just wanted to know how exactly Dagster decides which solids will be executed first. If they are independent, they should in run parallel, and not one by one, right?

geoHeil

01/05/2022, 3:33 PM

When working with dagster assets (and treating the metadata about them like a hive metastore which stores i.e. the path/location/db and table name) to the asset - how can I obtain the metadata for a given asset key? In https://dagster.slack.com/archives/C01U954MEER/p1640294374085400?thread_ts=1640209536.005500&cid=C01U954MEER @owen recommended

context.solid_def.input_defs[0].asset_key

which seems a bit strange/legacy and most importantly depending on the latest materialization of the asset (not considering its history). I would expect something like

a = context.assets['my_asset_key']

to get hold of all the metadata for the asset - and possibly then

a.location

to get the data about the location of the asset. Is this possible already? If not, what would the best way to get similar functionality to work?

Bryan Chavez

01/05/2022, 8:51 PM

I'm getting these errors below for sensor:

Copy code

@run_status_sensor(
    pipeline_run_status=PipelineRunStatus.SUCCESS,
    pipeline_selection=pipeline_names,
)

which seem strange since I can see the pipelines loaded in Dagit and the the pipeline for the sensor exists and can run. Any insight on what could be happening? SensorDaemon - ERROR - Sensor daemon caught an error for sensor job_success_email_sensor : dagster.core.errors.DagsterInvariantViolationError: No jobs, pipelines, graphs, or repositories found in "kh_dagster". Stack Trace: File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\grpc\server.py", line 205, in init self._repository_symbols_and_code_pointers.load() File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\grpc\server.py", line 90, in load self._loadable_repository_symbols = load_loadable_repository_symbols( File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\grpc\server.py", line 108, in load_loadable_repository_symbols loadable_targets = get_loadable_targets( File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\grpc\utils.py", line 37, in get_loadable_targets else loadable_targets_from_python_package(package_name) File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\core\workspace\autodiscovery.py", line 30, in loadable_targets_from_python_package return loadable_targets_from_loaded_module(module) File "C:\Users\<redacted>\Envs\<redacted>\lib\site-packages\dagster\core\workspace\autodiscovery.py", line 78, in loadable_targets_from_loaded_module raise DagsterInvariantViolationError(

Ryan Riopelle

01/05/2022, 11:00 PM

Whats a good way to specify an executable containing dagster pipelines using dagsterApiGrpcArgs in helm? I tried providing "--executable_path" as a parameter and it says its not an option.

Prratek Ramchandani

01/05/2022, 11:13 PM

i’m generating dagster jobs with a function (call in

make_job()

and am running into some trouble. i have a number of dagster jobs that vary only in resource config and so i have a python function that returns a

JobDefinition

to create them. since the graph for each job is identical i have one graph defined using the

@graph

decorator that gets calls in my

make_job()

function but dagster doesn’t like that i now have multiple graphs with the same name. i figured i’d name the graph using the “configured” API but looks like that requires that i also pass in config, which the graph doesn’t need. is there another way to name these graphs?

Bernardo Cortez

01/06/2022, 10:33 AM

I want to run a job after the successful run of another job. To do so, I am using the following pattern: 1. materializing assets in the last ops of the first job 2. a multi-asset sensor waits for those assets and fires the run of the second job Is there a better way to do this? I have looked into the documentation and, at first, I thought that run status sensors could be a solution, but then I couldn't fire the run a of a job in the sensor's decoration, like I would do in a @sensor decoration. Thanks!

Bastiaan

01/06/2022, 11:01 AM

Hi all, I have an op that needs to download several gigabytes of data which does not fit into memory. I tried solving this by yielding the data in batches as a dynamic output so that each batch can be processed sequentially (using max_concurrency=1) or distributed on multiple machines. The problem is that the job still seems to accumulate the generated data in memory. This occurs both when using the fs_io_manager or the s3_pickle_io_manager. Below is a code snippet illustrating the setup:

Copy code

@dagster.op(
    name='stream_data',
    config_schema={
        'data_feed_url': str,
        'batch_size': dagster.Field(int, default_value=500000),
    },
    out=dagster.DynamicOut(list)
)
def stream_data(context):
    with urlopen(context.op_config['data_feed_url']) as feed:
        objects = ijson.items(feed, 'item')
        for batch in enumerate(batch_iterator(
                objects,
                batch_size=context.op_config['batch_size']
        )):
            yield dagster.DynamicOutput(
                value=batch,
                mapping_key=f'data_{i}'
            )

I checked that running the generator in isolation (as a normal python function) does free up memory after yielding each batch. Is there something I can do to make sure each dynamic output is dereferenced after the io_manager has handled it, or is this not possible with dagster?

Daniel Suissa

01/06/2022, 11:32 AM

Howdy! Was hoping to get advice for a hack I'm trying to implement to get around the fact that versioning is not supported for dynamic DAGs. My ultimate goal is to prevent certain ops from running if their code didn't change and they ran within some time period (say the last day). In other words I'm trying to hack memoization until the issue is resolved. I was thinking: 1. in the beginning of the job save a code hash to S3 with the run id and a timestamp 2. in the beginning of my op get the hash from the latest timestamp and the run id 3. use the IO manager to get the results for the op with my run id 4. return the result in the op (if the hash / result isn't found the op would run the business logic again) Anything I should be alert of when doing something like this, or is there simpler way that leverages the IOManager instead of using an extra S3 bucket save? cc @Sa'ar Elias

Bernardo Cortez

01/06/2022, 12:09 PM

What would the best pattern be for passing information (like partition_dates, labels, etc...) from one job to another? Currently I am doing the following: 1. the last op of a job yields an asset materialization (AM) with the information to be passed to the next job 2. the sensor that is listening to these AMs reads that info and writes it in the resources of the run_config of the RunRequest 3. the next job's ops can access the info via

context.resources.(...)

George Pearse

01/06/2022, 2:13 PM

Can you use ipdb in dagster? e.g.

Copy code

context.ipdb.set_trace()

Jonathon Mobley

01/06/2022, 2:40 PM

Are there any recommendations or examples on how one could extend the redshift resource from the AWS library?

Bernardo Cortez

01/06/2022, 3:59 PM

how is the cursor of an

@asset_sensor

updated?

Alisson Gustavo

01/06/2022, 4:40 PM

👋 4

Cody Degen

01/06/2022, 10:14 PM

hey folks, fairly new user to dagster here, although my team has been using it for some time. I've intermittently been getting this error when running a pipelines that we have. The run will seem to fail, but best i can tell, it's because the sqlite database used to track events gets malformed, not because of any kind of error in the code itself, because i've run the same thing multiple times and gotten different results. This doesn't occur on other machines or for other users running the same pipelines. Error is below. This is for dagster version 0.11.12. I'm on windows 10, but running this using WSL. If there's any other information can provide, i'm happy to.

Copy code

Message: (sqlite3.DatabaseError) database disk image is malformed
[SQL: SELECT event_logs.id, event_logs.event 
FROM event_logs 
WHERE event_logs.run_id = ? ORDER BY event_logs.id ASC
 LIMIT ? OFFSET ?]
[parameters: ('f07ef76b-2e8e-44cb-990d-7b151f52462f', -1, 0)]
(Background on this error at: <http://sqlalche.me/e/14/4xp6>)

Path: 

Locations: 

Stack Trace:
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/rx/core/observablebase.py", line 67, in set_disposable
    subscriber = self._subscribe_core(auto_detach_observer)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/rx/core/anonymousobservable.py", line 20, in _subscribe_core
    return self._subscribe(observer)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/dagster_graphql/implementation/pipeline_run_storage.py", line 11, in __call__
    events = self.instance.logs_after(self.run_id, self.after_cursor)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/dagster/core/instance/__init__.py", line 997, in logs_after
    return self._event_storage.get_logs_for_run(run_id, cursor=cursor, of_type=of_type)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 199, in get_logs_for_run
    events_by_id = self.get_logs_for_run_by_log_id(run_id, cursor, of_type)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/dagster/core/storage/event_log/sql_event_log.py", line 166, in get_logs_for_run_by_log_id
    results = conn.execute(query).fetchall()
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1262, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 324, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1451, in _execute_clauseelement
    ret = self._execute_context(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1813, in _execute_context
    self._handle_dbapi_exception(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1994, in _handle_dbapi_exception
    util.raise_(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1770, in _execute_context
    self.dialect.do_execute(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 717, in do_execute
    cursor.execute(statement, parameters)

The above exception was the direct cause of the following exception:

Message: database disk image is malformed

Stack Trace:
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1770, in _execute_context
    self.dialect.do_execute(
  File "/mnt/c/Users/Cody/Documents/SimplyAnalytics/simplyanalytics-data-analysis/venv/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 717, in do_execute
    cursor.execute(statement, parameters)

Bryan Chavez

01/06/2022, 10:22 PM

Has anyone had any success deploying dagster/dagit to multi-container app on Azure?

Ryan Riopelle

01/07/2022, 12:06 AM

I noticed older version of dagster v0.13.0 does not provide volumeMounts as an option for helm user deployments. Is this correct? The alternative is envSecrets, which can be plainly provided and work, but when specifying valueFrom: name: key: does not work in both the old and new. Is this a bug?

Shreyas Karnik

01/07/2022, 12:08 AM

👋 Using dagster and loving it 😍! I have a question, is it possible to load a workspace dynamically via the api? The use case is that we want to run the core dagster services (dagit and daemon) and allow various teams to own the workspaces. Currently each time a new workspace is added have to manually edit the

workspace.yaml

file and reload dagit. All of our user code deployments run a grpc server. Is it possible to register new workspace and reload dagit via the API?

Roel Hogervorst

01/07/2022, 8:14 AM

I feel a bit dumb asking this question, but, if I have deployed a deamon, dagit, and a user code repository (with gRPC connection) . • On which 'image' is the usercode actually running? • Does it run on a different machine if I do a manual run of a job through the dagit interface or through a schedule? • And related, what information does each component (dagit, daemon, usercode repo) need to know? Do I have to supply AWS secrets to all?

geoHeil

01/07/2022, 11:55 AM

How can I prototype dagster functions in a jupyter (interactive) notebook? I.e. when I need a reference to an active dagit context:

context.solid_def.input_defs[0].asset_key

is it possible to somehow connect to a dagit instance running on localhost - and then exploring the API interactively?

Roel Hogervorst

01/07/2022, 12:40 PM

I have 2 ops that require inputs. How do I pass that through the config in job? I tried

Copy code

production_job = graphname.to_job(
   config= {"ops": {"op1":{"inputs":{"arg1":"bla", "ar2":"bla"}},
                    {"op2":{"inputs":{"arg1: "something"}
            })

But that doesn't work. It does work for 1 op, but I get

unexpected config entry

. I tried a list in stead of 2 dictionaries. But that is also not allowed. What is the best way to pass the arguments to the ops from the job?

Jessica Franks

01/07/2022, 12:57 PM

Hi! I'm running dagster in kubernetes using a helm chart and am trying to pass an envSecret to my sensor for slack alerts on failures, but it doesn't seem to be coming through envSecrets I've defined for other purposes in runLauncher work, so where/how are you supposed to define envSecrets for sensors? I've tried putting them in the dagit and daemon sections of the values file, but still not coming right. The sensor itself works when I define the token as a string value - so its just trying to get it from an envSecret that I am struggling with

Rodrigo Baron

01/07/2022, 1:00 PM

Hello, I've made a POC using Dagster and just loved it. Then after present to my team they have some questions about it: • We saw the support to run jobs on top of K8s, however this create one just one pod to run all tasks. That's correct? if so there's a way to run each task in a separated pod? • Regarding K8s, that's possible to run part of a job (like model training) using multiple pod's in parallel? • Yet about K8s 😅 there's a way to select the target node? let's say I have one high performance node and need select that one for the job (maybe using a label filter)..

Andy Chen

01/07/2022, 2:26 PM

Hey! Does dagster-pandas have the equivalent of “dtypes” in pandera? https://pandera.readthedocs.io/en/latest/dataframe_schemas.html#get-pandas-data-types I basically have a csv path that's an input to one of my ops, and I want to read that csv into a data frame. I want to explicitly define the types on its columns, but I don't want to define both the pandas-dagster schema and the dtype dictionary to pass into read_csv. In pandera, it looks like you can just define the schema once and then call .dtypes to pass into read_csv. Curious if you guys have a similar helper function or if I should just go with pandera?

raul

01/07/2022, 3:27 PM

Hello all. wanted to ask something basic. how do i list objects in GCS bucket ? Any sample code example i could get would help.