dagster #dagster-support

Alex Rudolph

03/03/2022, 6:12 PM

Hey all! A lot of our pipelines will occasionally fail with this error. We're on Dagster 0.13.12 and using the k8s_run_executor and are not sure why this started happening:

Copy code

kubernetes.client.exceptions.ApiException: (404)
Reason: Not Found
HTTP response headers: HTTPHeaderDict({'Audit-Id': '6829c683-9321-4b55-9592-c94794c50c81', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': 'd62db2cc-fd32-4618-af78-d12b3e721ec9', 'X-Kubernetes-Pf-Prioritylevel-Uid': 'f91e6c2b-ed56-4c59-98f2-420aed375018', 'Date': 'Thu, 03 Mar 2022 17:51:34 GMT', 'Content-Length': '282'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"jobs.batch \"dagster-job-365e6e829dfcac3a5f67eb4970876776-2\" not found","reason":"NotFound","details":{"name":"dagster-job-365e6e829dfcac3a5f67eb4970876776-2","group":"batch","kind":"jobs"},"code":404}



  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/api.py", line 775, in pipeline_execution_iterator
    for event in pipeline_context.executor.execute(pipeline_context, execution_plan):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/executor/step_delegating/step_delegating_executor.py", line 205, in execute
    plan_context, [step], active_execution
  File "/usr/local/lib/python3.7/site-packages/dagster_k8s/executor.py", line 217, in check_step_health
    job = self._batch_api.read_namespaced_job(namespace=self._job_namespace, name=job_name)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api/batch_v1_api.py", line 1257, in read_namespaced_job
    return self.read_namespaced_job_with_http_info(name, namespace, **kwargs)  # noqa: E501
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api/batch_v1_api.py", line 1366, in read_namespaced_job_with_http_info
    collection_formats=collection_formats)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 353, in call_api
    _preload_content, _request_timeout, _host)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 184, in __call_api
    _request_timeout=_request_timeout)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 377, in request
    headers=headers)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 243, in GET
    query_params=query_params)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 233, in request
    raise ApiException(http_resp=r)

Jay Sharma

03/03/2022, 9:22 PM

Hi all, I'm running dagster using k8s deployment and running it through minikube locally. My dagster version is 0.13.1. I'm getting a Crashloop error on one of the pods. When checking the logs for the error, I get this message: Error: No arguments given and workspace.yaml not found. I've used k8 deployment and minikube to run dagster locally successfully before. I've never had a workplace.yam filel so I'm not sure why this error is occurring now. To fix this, what arguments do I need to provide in the helm values.yaml file? Is this a helm issue, should I uninstall and re-install helm to fix this? Thanks for the help!

Will Gunadi

03/03/2022, 9:35 PM

Can two dagster-daemon processes run on one server (ubuntu)? The scenario is: one for Development and the other for Production

Stefan Adelbert

03/03/2022, 9:55 PM

Restart gRPC Repository When Code Has Changed I'm running

dagster

inside a docker container as part of my development workflow. I'd like to be able to modify job code and then be able to run that job code from

dagit

without having to rebuild the docker image and restart the repository container. My idea is to bind mount the job code into the container, so then I can edit the code without needing to rebuild the container. I already do this for running unit tests continuously using

pytest-watch

. Is there a way to cause

dagster api grpc --package-name <package>

to reload

<package>

on the fly?

Erik

03/04/2022, 12:28 AM

Hey I've noticed dagit starts absolutely crawling after maybe 10k jobs, it seems like its loading everything into memory in the partitions page. Any thoughts? Also are there docs for using the CLI with a remote deploy?

Chris Nogradi

03/04/2022, 12:48 AM

Is using a resource the best way to maintain state between each invocation of an op within a job?

Mark Fickett

03/04/2022, 1:38 AM

When multiple values are returned form an op, how does Dagster figure out which is which? Are they just sorted by output name or dict key order? In the conditional branching docs, there's this example. But if I set it to always yield

branch_1

but then assign

branch_2, branch_1 = branching_op()

in my

@job

, then I get backwards execution.

Copy code

# <https://docs.dagster.io/concepts/ops-jobs-graphs/jobs-graphs#conditional-branching>
import random

from dagster import Out, Output, job, op


@op(out={"branch_1": Out(is_required=False), "branch_2": Out(is_required=False)})
def branching_op():
    num = 0
    if num == 0:
        yield Output(1, "branch_1")
    else:
        yield Output(2, "branch_2")


@op
def branch_1_op(_input):
    pass


@op
def branch_2_op(_input):
    pass


@job
def branching():
    branch_2, branch_1 = branching_op()  # Does this always return branch_1, branch_2? Why?
    branch_1_op(branch_1)
    branch_2_op(branch_2)

Mark Fickett

03/04/2022, 3:22 AM

I'm having two issues in Dagit (0.13.9) with a large DAG (63 ops): • The page takes several minutes to load in waterfall view or timed view. (The job overview loads just fine.) If I prune 10% of the graph off, it works fine, and as I add more back in it gets way worse quickly, I'm wondering if there's an n^2 in a visual layout loop or something? It shows "Fetching logs" with a spinner while it's loading. I can load other Dagit pages just fine while the waterfall view is blank w/ a loading spinner, and Firefox or Chromium on Linux show me a 'the page is slowing down the browser' message. Same loading a running or completed job. • If I have scrolled down and to the right in the waterfall view (after it loads), when the graph status updates as ops complete, the view scrolls rapidly back up and to the left. Known issues? What do you need for a good bug report? My graph definition has a bunch of hooks into application logic that would be kind of a pain to strip out to send over an example, is there an intermediate DAG representation I could send?

Ben Gatewood

03/04/2022, 3:49 AM

Copy code

Operation name: JobMetadataQuery

Message: 'Select' object has no attribute 'filter'

Path: ["pipelineRunsOrError","results",0,"assets"]

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

Ben Gatewood

03/04/2022, 3:49 AM

Error suggested pasting it here so....here you go blob rofl

Ben Gatewood

03/04/2022, 3:50 AM

(0.14.3 btw)

Robert Schmidtke

03/04/2022, 12:36 PM

Hi, fairly new user here. Ran dagster+dagit locally, and now we want to adopt it on a broader scale, so I have a couple of general questions/wondering about best practices. Our idea is to deploy to ECS and spawn user tasks as needed, but willing to switch to another deployment scheme if need be. As I understand I have to tell dagit where the user code lives, either via workspace.yaml or by specifying command line arguments upon start. I was wondering what the scalable solution is to having many user code repositories. Do I have to update the workspace file every time that I add a new repository (say, load_from a new grpc connection)? Or is there a way for user agents to reach out to dagit and tell it about themselves? What I want to do is enable users to develop/add repositories on their own, while I IT/ops makes sure that dagit is up and running. Once the users are happy with their code, they push their image. I am wondering whether it's possible for that image to be started on ECS/k8s/... and connect to the long-running dagit instance, instead of extending the dagit's instance workspace.yaml with the new user code location (effectively requiring a restart?). Am I remotely on the right track here, or did I get some concepts completely backwards? Many thanks in advance!

👍 2

chrispc

03/04/2022, 1:54 PM

Hi Team (Hi, again @daniel), Could I get some help? I am trying to run dagster as multicontainer mode in my local machine. It was working with V0.12, but I migrated to V0.14 to test the new features, however, now It is not working. I am getting this error:

"File "/usr/local/lib/python3.9/site-packages/dagster_docker/docker_run_launcher.py", line 83, in _get_docker_image

raise Exception(“No docker image specified by the instance config or repository”)” , my docker-compose :

Copy code

version: "3.7"
services:
  dagster_pipelines:
    build:
      context: .
      dockerfile: ./Dockerfile_pipelines
    container_name: dagster_pipelines
    image: dagster_pipeline_image
    restart: always
    ports:
      - "4000:4000"
    env_file:
      - .env
    volumes:
      - ./-----.json:/var/-----.json
    networks:
      - dagster_network

  dagster_dagit:
    build:
      context: .
      dockerfile: ./Dockerfile_dagster
    entrypoint:
      - dagit
      - -h
      - "0.0.0.0"
      - -p
      - "3000"
      - -w
      - workspace.yaml
    container_name: dagster_dagit
    image: dagster_app
    expose:
      - "3000"
    ports:
      - "3000:3000"
    env_file:
      - .env
    volumes: # Make docker client accessible so we can terminate containers from dagit
      - /var/run/docker.sock:/var/run/docker.sock
    networks:
      - dagster_network
    depends_on:
      - dagster_pipelines
  dagster_daemon:
    entrypoint:
      - dagster-daemon
      - run
    container_name: dagster_daemon
    image: dagster_app
    restart: on-failure
    env_file:
      - .env
    volumes: # Make docker client accessible so we can launch containers using host docker
      - /var/run/docker.sock:/var/run/docker.sock
    networks:
      - dagster_network
    depends_on:
      - dagster_pipelines
networks:
  dagster_network:
    driver: bridge
    name: dagster_network

Dockerfile_pipelines:

Copy code

FROM python:3.9.10-slim
RUN pip install -U pip setuptools
RUN pip install --no-cache-dir\
    pandas==1.3.3 \
    dagster==0.14.2 \
    dagster-graphql==0.14.2 \
    dagit==0.14.2 \
    dagster-postgres==0.14.2 \
    dagster-docker==0.14.2 \
    dagster-gcp==0.14.2 \
    gcsfs==2022.2.0 \
    dateparser==1.0.0 \
    db-dtypes \
    great-expectations==0.14.4 \
    python-decouple==3.4 \
    avro


WORKDIR /var
COPY -------.json ./
ENV GOOGLE_APPLICATION_CREDENTIALS=/var/-----.json

WORKDIR /opt/dagster/app/
# Add repository code
RUN mkdir CAPS
RUN mkdir great_expectations_root
RUN mkdir -p local_artifact_storage/storage/

COPY CAPS/ CAPS/
COPY great_expectations_root/ great_expectations_root/
COPY config/local_docker/dagster.yaml config/local_docker/workspace.yaml setup.py ./

ENV DAGSTER_HOME=/opt/dagster/app/

RUN pip install -e .

# Run dagster gRPC server on port 4000

EXPOSE 4000

# Using CMD rather than ENTRYPOINT allows the command to be overridden in
# run launchers or executors to run other commands using this image
CMD ["dagster", "api", "grpc", "-h", "0.0.0.0", "-p", "4000", "-f", "CAPS/caps.py"]

Dockerfile_dagster:

Copy code

FROM python:3.9.10-slim

ENV DAGSTER_HOME=/opt/dagster/dagster_home/
ENV DAGSTER_PG_USERNAME=-----
ENV DAGSTER_PG_PASSWORD=-----
ENV DAGSTER_PG_HOST=----
ENV DAGSTER_PG_DB=---

RUN mkdir -p $DAGSTER_HOME

RUN pip install -U pip setuptools
RUN pip install --no-cache-dir\
    dagster==0.14.2 \
    dagster-graphql==0.14.2  \
    dagit==0.14.2  \
    dagster-postgres==0.14.2  \
    dagster-docker==0.14.2  \
    dagster-gcp==0.14.2

# Set $DAGSTER_HOME and copy dagster instance and workspace YAML there

COPY config/local_docker/dagster.yaml config/local_docker/workspace.yaml $DAGSTER_HOME

WORKDIR $DAGSTER_HOME

EXPOSE 3000

ENTRYPOINT ["dagit", "-h", "0.0.0.0", "-p", "3000"]

worskpace.yaml

Copy code

load_from:
  - grpc_server:
      host: dagster_pipelines
      port: 4000
      location_name: "dagster_pipelines"

dagster.yaml

Copy code

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

run_launcher:
  module: dagster_docker
  class: DockerRunLauncher
  config:
    env_vars:
      - DAGSTER_PG_USERNAME
      - DAGSTER_PG_PASSWORD
      - DAGSTER_PG_HOST
      - DAGSTER_PG_DB
    network: dagster_network
    container_kwargs:
      auto_remove: true

run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 25

run_storage:
  module: dagster_postgres.run_storage
  class: PostgresRunStorage
  config:
    postgres_db:
      username:
        env: DAGSTER_PG_USERNAME
      password:
        env: DAGSTER_PG_PASSWORD
      hostname:
        env: DAGSTER_PG_HOST
      db_name:
        env: DAGSTER_PG_DB
      port: 5432

Ant idea what is happening? Thank you in advance.

Mark Fickett

03/04/2022, 4:18 PM

Can I use the output of multiple

.collect()

calls in a order dependency

start

argument? Like this:

Copy code

@op(
  ins={"start": In(Nothing)}
)
def wait_for_many():
  pass

@gaph
def mostly_parallel():
   parallel_done = []
   parallel_done.append(has_dynamic_out.map(do_one_thing))
   parallel_done.append(has_dynamic_out.map(do_another_thing))
   parallel_done.append(has_dynamic_out.map(do_different_thing))
   wait_for_many(start=parallel_done)

When I try to do it as written above, I get:

DagsterInvalidDefinitionError: In @graph mostly_parallel, received a list containing an invalid type at index 0 for input "start" (passed by keyword) in op invocation wait_for_many. Lists can only contain the output from previous solid invocations or input mappings, received <class 'list'>

(actual op names changed to match my minimal example).

Alex Service

03/04/2022, 6:06 PM

When looking at

storage_id

, is it guaranteed to be monotonically increasing? Sequential?

Danny Jackowitz

03/04/2022, 8:44 PM

👋 Question regarding using cursors with sensors as described here: https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#sensor-optimizations-using-cursors What are the transactional guarantees around

update_cursor()

and yielded run requests? Is there anything we need to watch out for that could result in missed runs while using cursors (e.g. the cursor is updated but the run never actually happens)? Or is it guaranteed that the cursor update will never be recorded without the run request also?

Alexander Verbitsky

03/04/2022, 8:44 PM

Hi, All. What is mean, when run is in

Started

status, but in the same time all tasks have already finished?

Will Gunadi

03/05/2022, 12:05 AM

Is there an equivalent of context.scheduled_execution_time for jobs that are not run via the scheduler? Context: When I run a job manually via dagit, the above variable don't exist in the context.

Anoop Sharma

03/05/2022, 9:42 AM

Is there a way to have a shared variable available across all the ops within a job, whose value can also be updated by a certain op in the job? Essentially, an editable environment variable for a job.

sashank

03/05/2022, 6:11 PM

Running into these errors while trying to use dynamic mapping and collect. Any ideas about what I’m doing wrong?

Copy code

@graph
def my_graph():
    dynamic_op().map(op_with_input)

> AttributeError: 'InvokedSolidOutputHandle' object has no attribute 'map'


@graph
def my_graph():
    dynamic_op.map(op_with_input)

> AttributeError: 'OpDefinition' object has no attribute 'map'

sashank from elementl inc 1

Tyler Hillery

03/05/2022, 11:56 PM

If I have dagster-daemon running in the background on my local machine do I also need to have dagit running for a scheduled job to kick off? According the docs I don't think so but wanted some clarification. This is the part of the docs I am referring to ➡️

This process will periodically check for any running schedules and launch their associated runs. If you leave this process running, it will launch a new run for your schedule each day at the expected time.

jasono

03/06/2022, 7:36 AM

Hi,

dagster-ge

seems to support the old version of Great Expectations (config version 2). Any idea when it will be updated to v3?

efrat

03/06/2022, 8:20 AM

Hi, Is there a detailed example of how to create a resource for connecting to a database (mySql)?

geoHeil

03/06/2022, 4:23 PM

Does Dagster have something like an SFTP sensor i.e. like https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.6.0/org.apache.nifi.processors.standard.GetFTP/ for NiFi? https://docs.dagster.io/_apidocs/libraries/dagster-ssh seems to be quite bare-bones for connectivity - but not a sensor

jasono

03/06/2022, 4:38 PM

Hi, I keep getting this

invalid argument: dagster_home\\logs\\event.log'

error when running

dagit -f hello_cereal.py

. I don’t get any error if it’s run by

python hello_cereal.py

. Can someone please take a look at the error message below? If I try to read

event.log

file manually, I get an unknown file system error.

Stefan Adelbert

03/06/2022, 10:38 PM

Mixing order and data based dependency If I have a graph like below: A -> B -> C where C needs to execute after B, but C gets its data from A, how do I use a combination of order and data based dependencies to achieve this? If I try something like,

Copy code

@op(ins={"start": In(Nothing), "data": In(Any)})
def C(context, start, data):
   # do stuff

I get the following error

Copy code

dagster.core.errors.DagsterInvalidDefinitionError: @op 'C' decorated function has parameter 'start' that is one of the input_defs of type 'Nothing' which should not be included since no data will be passed for it.

Any tips?

Stefan Adelbert

03/07/2022, 5:25 AM

Sensor that uses resources I'd like to create a sensor that triggers a job for each email in an inbox matching certain criteria. I have already created a resource which my ops use to interact with a gmail account. That gmail resource is itself dependent on another resource which supplies authentication credentials. I'd like to reuse these resources with my sensor, but it seems like

@sensor

doesn't support providing resources. Any tips on reusing resources with sensors?

Max

03/07/2022, 7:14 AM

what's the recommended strategy to build a dagit container with a single workspace and a single repo? My current setup is •

DAGSTER_HOME=/opt/dagster/dagster_home

. Copy the dagster.yml there (works just fine) • workdir is

/opt/dagster/app

. I copy all my files there, including

/opt/dagster/app/my_project/repository.py

and the

workspace.yaml

in the root • my workspace contains

python_file: relative_path: my_project/repository.py

. This loads correctly when I run

dagit

locally but not in docker (

module my_project was not found

) What's the best practice there to make sure that the import works?

Irven Aelbrecht

03/07/2022, 7:35 AM

Are there any plans to support

Ray

? (https://github.com/ray-project/ray) and for my understanding was there any reason to choose

Dask

over

Ray

for multi-processing?

Chen Tsinovoy

03/07/2022, 8:25 AM

Hey ! I want to mock a resource I created, so I used magic mock as following:

Copy code

from unittest.mock import MagicMock
mocked_http = http_async_downloader()
mocked_http.fetch_all = MagicMock(return_value=["4 /n hello"])


@job(resource_defs={"http": mocked_http})
def mocked_job():
    print_htmls(download_urls(convert_cids_to_urls()))


if __name__ == "__main__":
    # upload_pubchem_3d_conformers_dev_job.execute_in_process()
    mocked_job.execute_in_process(run_config=config_from_files([file_relative_path(__file__, 'config.yaml')]))

but I got the following exception:

Copy code

dagster.check.CheckError: Value in dictionary mismatches expected type for key http. Expected value of type <class 'dagster.core.definitions.resource_definition.ResourceDefinition'>. Got value <resources.aka_http.HttpAsyncDownloader object at 0x7fd8b05cc390> of type <class 'resources.aka_http.HttpAsyncDownloader'>.

what am I missing ?