dagster #dagster-support

Anaqi Afendi

10/28/2021, 3:41 PM

How can I get the output of a notebook in dagster through dagstermill? I've followed the guide and API docs on the site but I keep running into this errror:

Copy code

dagster.core.errors.DagsterExecutionStepExecutionError: Error occurred while executing solid "dbt_job":
...

The above exception was caused by the following exception:
KeyError: ('55fafb57-29ab-4d0e-a7b2-c6a30bcdf1db', 'dbt_job', 'dbt_job_status')
  File "C:\Users\aafendi\Anaconda3\envs\dagster-environment\lib\site-packages\dagster\core\execution\plan\utils.py", line 44, in solid_execution_error_boundary
    yield
  File "C:\Users\aafendi\Anaconda3\envs\dagster-environment\lib\site-packages\dagster\utils\__init__.py", line 383, in iterate_with_context
    next_output = next(iterator)
  File "C:\Users\aafendi\Anaconda3\envs\dagster-environment\lib\site-packages\dagstermill\solids.py", line 268, in _t_fn
    build_input_context(upstream_output=output_context)
  File "C:\Users\aafendi\Anaconda3\envs\dagster-environment\lib\site-packages\dagster\core\storage\mem_io_manager.py", line 14, in load_input
    return self.values[keys]

Slackbot

10/28/2021, 5:32 PM

This message was deleted.

Anaqi Afendi

10/28/2021, 9:44 PM

I have a jupyter notebook running in dagstermill, but it imports from another module that I wrote. I am getting an error that the module cannot be found. I have both the module and the notebook in a "modules" folder, and the modules folder is in the same dir as the executing dagster python file. Is this just a limitation of dagstermill currently or is there a way around this?

Louis Auneau

10/29/2021, 8:14 AM

Hi everyone! I am trying Dagster for the first time and I have a question that is more about “best practises”. We have a very simple “job” that calls a function

get_model(name: str)

twice to then do more complex stuff. In our original project we defined the names of our models as constants and then called the functions using those constants

get_model(MY_MODEL_NAME)

. However when I transform my function into an op/solid, it doesn’t work since my constants are not another op/solid’s output. What’s the best practise to handle such cases ? Should I create one op/solid per constant which is just a function returning the value ? Thank you by advance and have an excellent day !

Jori Geysen

10/29/2021, 12:46 PM

Hi all, I'm looking to include python standard logs with dagit logs. I've gone through the documentation for dagster logging. My overall implementation details are the following: • dagster 0.13.1 • multi-container docker deployment as described here ◦ including a dagster-daemon container ◦ including a dagster gRPC service for user code As I mentioned, I'm looking to include standard python logs with dagit logs. I came up with two configurations for the

dagster.yaml

file: 1. When I want to save all logs to a file:

Copy code

python_logs:
  managed_python_loggers:
    - my_logger
  dagster_handler_config:
    handlers:
      console:
          class: logging.FileHandler
          level: DEBUG
          stream: "logfile.log"

This works great, both the logs from

mylogger

and the dagster logs from the user code are saved to

logfile.log

, sitting in the gRPC container. 2. When I want to view the logs in terminal by running

docker logs

on the gRPC container (without saving to file first):

Copy code

python_logs:
  managed_python_loggers:
    - my_logger
  dagster_handler_config:
    handlers:
      console:
          class: logging.StreamHandler
          level: DEBUG
          stream: <ext://sys.stdout>

This does not work however. When running

docker logs

for the gRPC container, only the the dagster logs from the user code logs show up. The logs emitted by

mylogger

do however show up on the dagit UI. => where are the 'standard' python logs direct to when the StreamHandler is used in combination with
<ext://sys.stdout>
? Thanks a lot for your help in advance, have a nice day !

Andy H

10/29/2021, 3:47 PM

I am receiving an error regarding run configurations after moving to

0.13.2

and changing to

ops

from

solids

. My run config appears to be correct, any ideas why I'm getting this exception?

Sandeep Devarapalli

10/29/2021, 4:46 PM

Hi Team, how can I use dagster with clickhouse?

Alexander Butler

10/29/2021, 9:19 PM

Hello All, If I am using map collect with an op that has three outputs I would like to pipe to 3 different functions. Looks like the example image. Has anyone had experience trying this successfully? I feel like I am close.

Marcel M

10/31/2021, 7:52 PM

Pipeline newbie here. I could use some help with the following problem. I’m using the new API. I have a set of database tables that I want to read into a Pandas Dataframe, and then write each dataframe to another SQL database in a table for each dataframe. (I will add some more transformation later). So in fact a mini pipeline repeated for a list of tuples of (name, query). Sounds simple, but I could not get it to work. It seemed that I needed a Dynamic mapping and “call” the pipeline for each tuple. I implemented the pipeline as a graph which I want to map. My code is something like this:

Copy code

@op(out={'spec': DynamicOut(Tuple[str, str])})
def gen_query_specs():
    queries = [
        ('activiteit', 'SELECT * FROM activiteit_data'),
        ('afsluitreden', 'SELECT * FROM afsluitreden_data'),
        ('aanleiding', 'SELECT * FROM aanleiding_data'),
    ]
    for q in queries:
        yield DynamicOutput(
            output_name='spec',
            value=q,
            mapping_key=q[0],
        )

Copy code

@op(required_resource_keys={"openac"})
def ingest_query(context, query: str):
    <http://context.log.info|context.log.info>(f"Running query on OpenAC: {query}")
    df = context.resources.openac.query(query)
    return df


@op(required_resource_keys={"rel_wh"})
def save_to_wh(context, df: pd.DataFrame, df_name: str):
    df = context.resources.rel_wh.save_df(df_name, df)
    <http://context.log.info|context.log.info>(f"Saved '{df_name}' to RelDB")


@graph
def ingest(spec: Tuple[str, str]):
    # <http://context.log.info|context.log.info>(f"Ingesting {spec[0]}")    # context not available?
    query = spec[1] # <== error here: IndexError: tuple index out of range 
    df = ingest_query(query)
    df_name = spec[0]
    save_to_wh(df, df_name)
    # <http://context.log.info|context.log.info>(f"Done ingesting {spec[0]}")

@graph
def ingest_openac():
    specs = gen_query_specs()
    specs.map(ingest)



job = ingest_openac.to_job(
    resource_defs={
        "openac": openac_service,
        "rel_wh": reldb_warehouse_service,
    },
    config=config_from_files(["resources.yaml"]),
)


if __name__ == "__main__":
    result = ingest_openac.execute_in_process()

Is this the wrong way to approach this problem? What is the best way to handle this kind of repetition? How does one map more than one op to a dynamic mapping?

Francis

11/01/2021, 1:57 AM

hello, I’d like my schedule to not run a new pipeline if a current run is ongoing. I’m not sure how I can configure that based on the docs. if it exists, could someone fill me in or if not, how might one implement this functionality. thanks!

👍 1

Kateryna Chenina

11/01/2021, 11:59 AM

Hi everyone! After transitioning to the new jobs, ops API, I can't run complex pipelines. If I add only one graph to the Job, it all executes fine. But if I include two and more, I get this error:

Copy code

dagster.core.executor.child_process_executor.ChildProcessCrashException
  File "c:\users\kateryna\documents\pythonscripts\comp_rate_predictor\venv\lib\site-packages\dagster\core\executor\multiprocess.py", line 163, in execute
    event_or_none = next(step_iter)
  File "c:\users\kateryna\documents\pythonscripts\comp_rate_predictor\venv\lib\site-packages\dagster\core\executor\multiprocess.py", line 268, in execute_step_out_of_process
    for ret in execute_child_process_command(command):
  File "c:\users\kateryna\documents\pythonscripts\comp_rate_predictor\venv\lib\site-packages\dagster\core\executor\child_process_executor.py", line 158, in execute_child_process_command
    raise ChildProcessCrashException(exit_code=process.exitcode)

Multiprocess executor: child process for step DS2_1_UCP.get_process_ucp_date_of_inquiry unexpectedly exited with code 3221225477

Jeremy Fisher

11/01/2021, 9:36 PM

Anyone else having an issue adding a configurable dagstermill op?

Matthew Smicker

11/01/2021, 11:54 PM

I am upgrading to the new core api and making use of nested graphs to abstract away some complexity. I have a fairly simple case with the first op taking 2 strings and outputting a string. the second op (within the graph) takes the string as input and outputs a python class (which is defined as the dagster type). The step_output type check passes but then I get this error message:

Copy code

TypeError: cannot pickle '_thread.lock' object
  File "/usr/local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_plan.py", line 195, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/usr/local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_step.py", line 327, 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.8/site-packages/dagster/core/execution/plan/execute_step.py", line 381, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output, input_lineage):
  File "/usr/local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_step.py", line 491, in _store_output
    handle_output_res = output_manager.handle_output(output_context, output.value)
  File "/usr/local/lib/python3.8/site-packages/dagster/core/storage/fs_io_manager.py", line 122, in handle_output
    pickle.dump(obj, write_obj, PICKLE_PROTOCOL)

Any ideas?

Arturs Stramkals

11/01/2021, 11:57 PM

I have a question along the lines of Matthew’s inquiry - was there some underlying parallel processing change in .13 release? My workflow starts with database connection generation, i.e. an op that outputs SQLAlchemy’s

create_engine

output, that now fails to pickle. I think I know how I can move this around, however.

Max Wong

11/02/2021, 7:24 AM

got a

search redirect

bug I think steps: 1. press

to search for a pipeline name 2. it redirects me to

<http://localhost:3000/workspace/crm@repos.py/jobs/crm>

--> produces

Unknown repository

error but the actual path is

<http://localhost:3000/workspace/baania_repository@repos.py/jobs/crm>

this is my workspace config:

Francis

11/02/2021, 11:39 AM

hello! got this error when executing a pipeline. Not sure what the source might be:

Copy code

Execution of pipeline [x] failed. An exception was thrown during execution.
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) database is locked
[SQL: INSERT INTO event_logs (run_id, event, dagster_event_type, timestamp, step_key, asset_key, partition) VALUES (?, ?, ?, ?, ?, ?, ?)]

Context, • We’re running multiprocessing on multiple concurrent pipelines • This looks like it happens almost at random

Daniel Salama

11/02/2021, 12:04 PM

Hey guys, is there a way to get the run id of the current run ? for example, after i create a new run :

Copy code

yield RunRequest(
    run_key="somename-{}".format(time_stamps),
    run_config={"solids": {"somename_step1": {"config": {"filename": "file"}}}},
)

can i get the run id that was created during the run ? i want to print logs to third party platform with the runid during the run .

Nick Dellosa

11/02/2021, 12:38 PM

Hey dagster team - is there a way to pass a Kubernetes config map of env variables to the

pipelineRun

configuration? It doesn't look there's an option for it in the helm chart: https://github.com/dagster-io/dagster/blob/master/helm/dagster/values.yaml#L298

dagster bot resolve to discussion 1

Nathan Saccon

11/02/2021, 7:09 PM

Hello! version = dagster 0.13.3 When I try and run this op:

Copy code

@op
def recreate_error() -> tuple[str, list[str]]:
    return ("str1", ["str2", "str3", "str4"])

I get this error:

dagster.core.errors.DagsterTypeCheckDidNotPass: Type check failed for step output "result" - expected type "Tuple[String,[String],String,String,String]".

Is this the correct behavior here? This seems incorrect

Koby Kilimnik

11/03/2021, 2:28 PM

i managed to install dagster and run dags, but how do i run them remotely on my ecs?

Koby Kilimnik

11/03/2021, 2:28 PM

cant find a good ecs runner tutorial

Koby Kilimnik

11/03/2021, 2:28 PM

dagster is installed on a single ec2 machine

Nicolas Gaillard

11/03/2021, 4:03 PM

Hi, is it possible to reload a workspace from dagit (meaning trigger the function with the decorator

@repository

in order to redeploy it) ? I thought the "reload" button was for that but it seems not.

Kirk Stennett

11/03/2021, 4:30 PM

Hey all, how do I enable retries using dagster-k8s and the k8s job executor? I'm getting errors that it's not enabled even though retries are requested.

Will Gunadi

11/03/2021, 5:28 PM

I may just miss something simple, but is there a way to save a Launchpad run, so I don't have to keep typing in the necessary "resources" that the run needs?

Chris Chan

11/03/2021, 7:30 PM

Is there guidance on how much memory the

dagster-daemon

uses? Our deployment had allocated 1.5 GB and it’s constantly OOMing (which isn’t a problem since the all the schedules and sensors run fine) but I’m worried that memory usage scales with the number of schedules and sensors

Makoto

11/03/2021, 8:46 PM

I was checking out

dagster_pandas

, and trying to figure out a good way to show which column is in violation of the constraints you specified. Is there an example somewhere? Just showing that the DataType failed is a bit hard to debug.

Andy Chen

11/03/2021, 9:16 PM

Hi, is there any documentation what dagster decides is the "root" directory? I've seen references to working_directory in workspace.yaml but I don't think that's it. I added a line to log getcwd() into one of my ops:

Copy code

@op
def sort_branches(context):
    <http://context.log.info|context.log.info>("hello " + os.getcwd())
    branches_df = pd.read_csv("pipeline/pjm/raw/branches.csv")
    return branches_df.sort_values(by=['branch_id'])

the path changes depending on whether its run from dagster cli, or in dagit. Dagit launch execution logs: "hello /Users/andychen/dev/nira-data-pipeline/pipeline"

Copy code

dagster job execute -f pipeline/pipeline/pjm/graphs/pjm_graph.py

logs: "sort_branches - hello /Users/andychen/dev/nira-data-pipeline" This is messing stuff up because I need to read csv's from a particular file path in my project directory, but depending on how I run it, it can't find the file. workspace.yaml:

Copy code

# Workspaces specify where to find user code in order to load it for Dagit and the Dagster CLI.
# See our documentation overview on Workspaces: <https://docs.dagster.io/overview/repositories-workspaces/workspaces>
load_from:
  - python_package: pipeline

Arturs Stramkals

11/03/2021, 9:33 PM

I appear to be missing something about contexts. How do I combine a context and normal arguments within an op’s function signature? I guess the question is on what to supply to an op with a signature

def foo_op(context, bar)

Xiaohui Liu

11/04/2021, 1:43 AM

I just started to use Dagster recently. I deployed dagster using the helm package (0.13.2). K8S isolated/stepped jobs are all running fun. But when I configured it to use celery-k8s, I got errors from dagit, It seems dagit could not find a config key called ‘celery-k8s’. Has anyone seen this error before?