Hey @Jonathan Ming - mind posting this in #dagster-cloud channel?

is there a possibility of adding count of rows as they are being processed into the dagit ui? kinda like with talend if you would process rows, it counts them and gives you a row/sec number

hi guys. I want to ask, how to do dbt cli command like "$ dbt run --full-refresh --select my_incremental_model+" in dagster as a job so we can run it to full refresh the "my_incremental_model" ???

Hi team! I'd like to say that I really like the new assets-related features! That's amazing! Thanks you for your great job! I'm in process of migration from ops to assets and some interesting question appeared: were there any attempts to represent DB schema migrations as versioned assets? I'm thinking about Alembic, but it seems too complex from the first look.

Hi everyone, I have pipelines stuck in

STARTING

mode. They keep on accumulating over the period of time. Currently I have to terminate them manually from DAGIT UI. I want to terminate them programmatically using the Graphql api. Is there a way this can be achieved? I am using

pipelineRunsOrError

query to filter out the ones that are in

STARTING

mode, but I can't tell how long they have been in that state. I would like to terminate the ones that pending for more than a day.

Hi, I have the following problem: I've created DBT project with Redshift, and I wanted to create dwonstream asset (plot) from the upstream DBT asset. When I execute my code, Dagster tries to read asset from the local folder, and not the Redshift table. I can't find references in docs on how to tackle this, since the examples use local DuckDB. Anyone know how to solve this? My error message is: FileNotFoundError:

[Errno 2] No such file or directory: '/repo/dagster_redshift_demo/tmp2sdfgg3/storage/dagster_demo_schema/customers'

I defined my asset as in tutorial https://docs.dagster.io/integrations/dbt/using-dbt-with-dagster/part-four :

@asset(
    ins={"customers": AssetIn(key_prefix=["dagster_demo_schema"])},
    group_name="staging",
)

Hi all,

Hi Guys,

Hi Guys,

Hi Guys, After upgrading to dagster 1.x, none of my jobs run, getting the following error:

TypeError: run() got an unexpected keyword argument 'remove'
  File "/usr/local/lib/python3.7/site-packages/dagster/_daemon/run_coordinator/queued_run_coordinator_daemon.py", line 330, in _dequeue_run
    LaunchRunContext(pipeline_run=run, workspace=workspace)
  File "/usr/local/lib/python3.7/site-packages/dagster_docker/docker_run_launcher.py", line 153, in launch_run
    self._launch_container_with_command(run, docker_image, command)
  File "/usr/local/lib/python3.7/site-packages/dagster_docker/docker_run_launcher.py", line 107, in _launch_container_with_command
    **container_context.container_kwargs,
  File "/usr/local/lib/python3.7/site-packages/docker/models/containers.py", line 905, in create
    create_kwargs = _create_container_args(kwargs)
  File "/usr/local/lib/python3.7/site-packages/docker/models/containers.py", line 1123, in _create_container_args
    raise create_unexpected_kwargs_error('run', kwargs)

I have went through the migration documentation and executed steps to no avail. Appreciate your support.

I’m getting the below error when I try to install dagit My setup: Mac M2 Using remote development via docker desktop I’ve tried both:

"image": "<http://mcr.microsoft.com/devcontainers/python:0-3.9-bullseye|mcr.microsoft.com/devcontainers/python:0-3.9-bullseye>",

"image": "<http://mcr.microsoft.com/devcontainers/python:0-3.10-bullseye|mcr.microsoft.com/devcontainers/python:0-3.10-bullseye>",

vscode ➜ /workspaces/a-api-integration (setup ✗) $ pip install dagit
Defaulting to user installation because normal site-packages is not writeable
Collecting dagit
  Using cached dagit-1.1.9-py3-none-any.whl (6.3 MB)
Collecting uvicorn[standard]
  Using cached uvicorn-0.20.0-py3-none-any.whl (56 kB)
Collecting dagster==1.1.9
  Using cached dagster-1.1.9-py3-none-any.whl (1.1 MB)
Collecting click<9.0,>=7.0
  Using cached click-8.1.3-py3-none-any.whl (96 kB)
Collecting watchdog>=0.8.3
  Using cached watchdog-2.2.1-py3-none-manylinux2014_aarch64.whl (78 kB)
Collecting PyYAML
  Using cached PyYAML-6.0-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl (731 kB)
Collecting requests
  Using cached requests-2.28.1-py3-none-any.whl (62 kB)
Collecting starlette
  Using cached starlette-0.23.1-py3-none-any.whl (64 kB)
Collecting dagster-graphql==1.1.9
  Using cached dagster_graphql-1.1.9-py3-none-any.whl (134 kB)
Collecting tabulate
  Using cached tabulate-0.9.0-py3-none-any.whl (35 kB)
Collecting universal-pathlib
  Downloading universal_pathlib-0.0.21-py3-none-any.whl (25 kB)
Collecting Jinja2
  Downloading Jinja2-3.1.2-py3-none-any.whl (133 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 133.1/133.1 KB 888.7 kB/s eta 0:00:00
Collecting sqlalchemy>=1.0
  Downloading SQLAlchemy-1.4.46-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl (1.6 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 1.6/1.6 MB 935.1 kB/s eta 0:00:00
Collecting packaging<22,>=20.9
  Using cached packaging-21.3-py3-none-any.whl (40 kB)
Collecting pydantic
  Downloading pydantic-1.10.4-py3-none-any.whl (155 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 155.3/155.3 KB 979.7 kB/s eta 0:00:00
Requirement already satisfied: setuptools in /usr/local/lib/python3.9/site-packages (from dagster==1.1.9->dagit) (58.1.0)
Collecting protobuf<4,>=3.13.0
  Using cached protobuf-3.20.3-cp39-cp39-manylinux2014_aarch64.whl (918 kB)
Collecting typing-extensions>=4.0.1
  Downloading typing_extensions-4.4.0-py3-none-any.whl (26 kB)
Collecting coloredlogs<=14.0,>=6.1
  Downloading coloredlogs-14.0-py2.py3-none-any.whl (43 kB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 43.9/43.9 KB 5.2 MB/s eta 0:00:00
Collecting python-dotenv
  Using cached python_dotenv-0.21.0-py3-none-any.whl (18 kB)
Collecting tqdm
  Using cached tqdm-4.64.1-py2.py3-none-any.whl (78 kB)
Collecting alembic!=1.6.3,!=1.7.0,>=1.2.1
  Using cached alembic-1.9.1-py3-none-any.whl (210 kB)
Collecting grpcio<1.48.1,>=1.32.0
  Downloading grpcio-1.47.2-cp39-cp39-manylinux_2_17_aarch64.whl (62.2 MB)
     ━━━╺━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 5.2/62.2 MB 953.7 kB/s eta 0:01:00
ERROR: Exception:
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/urllib3/response.py", line 438, in _error_catcher
    yield
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/urllib3/response.py", line 519, in read
    data = self._fp.read(amt) if not fp_closed else b""
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/cachecontrol/filewrapper.py", line 90, in read
    data = self.__fp.read(amt)
  File "/usr/local/lib/python3.9/http/client.py", line 463, in read
    n = self.readinto(b)
  File "/usr/local/lib/python3.9/http/client.py", line 507, in readinto
    n = self.fp.readinto(b)
  File "/usr/local/lib/python3.9/socket.py", line 704, in readinto
    return self._sock.recv_into(b)
  File "/usr/local/lib/python3.9/ssl.py", line 1242, in recv_into
    return self.read(nbytes, buffer)
  File "/usr/local/lib/python3.9/ssl.py", line 1100, in read
    return self._sslobj.read(len, buffer)
socket.timeout: The read operation timed out

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/cli/base_command.py", line 167, in exc_logging_wrapper
    status = run_func(*args)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/cli/req_command.py", line 205, in wrapper
    return func(self, options, args)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/commands/install.py", line 339, in run
    requirement_set = resolver.resolve(
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/resolver.py", line 94, in resolve
    result = self._result = resolver.resolve(
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/resolvers.py", line 481, in resolve
    state = resolution.resolve(requirements, max_rounds=max_rounds)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/resolvers.py", line 373, in resolve
    failure_causes = self._attempt_to_pin_criterion(name)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/resolvers.py", line 213, in _attempt_to_pin_criterion
    criteria = self._get_updated_criteria(candidate)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/resolvers.py", line 204, in _get_updated_criteria
    self._add_to_criteria(criteria, requirement, parent=candidate)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/resolvers.py", line 172, in _add_to_criteria
    if not criterion.candidates:
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/resolvelib/structs.py", line 151, in __bool__
    return bool(self._sequence)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/found_candidates.py", line 155, in __bool__
    return any(self)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/found_candidates.py", line 143, in <genexpr>
    return (c for c in iterator if id(c) not in self._incompatible_ids)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/found_candidates.py", line 47, in _iter_built
    candidate = func()
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/factory.py", line 215, in _make_candidate_from_link
    self._link_candidate_cache[link] = LinkCandidate(
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/candidates.py", line 288, in __init__
    super().__init__(
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/candidates.py", line 158, in __init__
    self.dist = self._prepare()
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/candidates.py", line 227, in _prepare
    dist = self._prepare_distribution()
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/resolution/resolvelib/candidates.py", line 299, in _prepare_distribution
    return preparer.prepare_linked_requirement(self._ireq, parallel_builds=True)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/operations/prepare.py", line 487, in prepare_linked_requirement
    return self._prepare_linked_requirement(req, parallel_builds)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/operations/prepare.py", line 532, in _prepare_linked_requirement
    local_file = unpack_url(
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/operations/prepare.py", line 214, in unpack_url
    file = get_http_url(
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/operations/prepare.py", line 94, in get_http_url
    from_path, content_type = download(link, temp_dir.path)
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/network/download.py", line 146, in __call__
    for chunk in chunks:
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/cli/progress_bars.py", line 304, in _rich_progress_bar
    for chunk in iterable:
  File "/usr/local/lib/python3.9/site-packages/pip/_internal/network/utils.py", line 63, in response_chunks
    for chunk in response.raw.stream(
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/urllib3/response.py", line 576, in stream
    data = self.read(amt=amt, decode_content=decode_content)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/urllib3/response.py", line 541, in read
    raise IncompleteRead(self._fp_bytes_read, self.length_remaining)
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/pip/_vendor/urllib3/response.py", line 443, in _error_catcher
    raise ReadTimeoutError(self._pool, None, "Read timed out.")
pip._vendor.urllib3.exceptions.ReadTimeoutError: HTTPSConnectionPool(host='<http://files.pythonhosted.org|files.pythonhosted.org>', port=443): Read timed out.
WARNING: You are using pip version 22.0.4; however, version 22.3.1 is available.
You should consider upgrading via the '/usr/local/bin/python -m pip install --upgrade pip' command.
vscode ➜ /workspaces/a-api-integration (setup ✗) $

Hi Dagster, we’ve just set up a multi_asset_sensor but we’re getting this error, see thread 🧵

Hi, has anyone observed any issues when scaling up User Code Deployments? After scaling my UC Deployment to 3 replicas i observe following: • In Dagit, i Preiodically get

Definitions Reloaded

popup. • In Dagit I also I observe error message:

Enum 'LocationStateChangeEventType' cannot represent value: <LocationStateChangeEventType instance>

• Dagit becomes less responsive • Also I notice slow linear increase of memory footprint of

dagit

and

daemon

pods (Helm Deployment v 1.1.9, tested in 2 different clusters -> Azure & Rancher Desktop)

Hi Everyone! I would like to ask about dagster-snowflake library. Here is the documentation for this topic: https://docs.dagster.io/_apidocs/libraries/dagster-snowflake Here is the example code what I used for snowflake resource connection:

from dagster import job, op
from dagster_snowflake import snowflake_resource

@op(required_resource_keys={'snowflake'})
def get_one(context):
    context.resources.snowflake.execute_query('SELECT 1')

@job(resource_defs={'snowflake': snowflake_resource})
def my_snowflake_job():
    get_one()

my_snowflake_job.execute_in_process(
    run_config={
        'resources': {
            'snowflake': {
                'config': {
                    'account': {'env': 'SNOWFLAKE_ACCOUNT'},
                    'user': {'env': 'SNOWFLAKE_USER'},
                    'password': {'env': 'SNOWFLAKE_PASSWORD'},
                    'database': {'env': 'SNOWFLAKE_DATABASE'},
                    'schema': {'env': 'SNOWFLAKE_SCHEMA'},
                    'warehouse': {'env': 'SNOWFLAKE_WAREHOUSE'},
                }
            }
        }
    }
)

and I’d like to use private key as in this Snowflake documentation: https://docs.snowflake.com/en/user-guide/python-connector-example.html#label-python-key-pair-authn-rotation

import snowflake.connector
import os
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.hazmat.primitives import serialization
with open("<path>/rsa_key.p8", "rb") as key:
    p_key= serialization.load_pem_private_key(
        key.read(),
        password=os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
        backend=default_backend()
    )

pkb = p_key.private_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption())

ctx = snowflake.connector.connect(
    user='<user>',
    account='<account_identifier>',
    private_key=pkb,
    warehouse=WAREHOUSE,
    database=DATABASE,
    schema=SCHEMA
    )

cs = ctx.cursor()

but when I have tried to use in the above dagster code the private key parameter, I have got this error message:

dagster._core.errors.DagsterInvalidConfigError: Error in config for resource snowflake

Error 1: Received unexpected config entry "private_key" at path root:config.

Error 2: Missing required config entry "password" at path root:

Does the dagster-snowflake library ready for private key authentication? Thank your in advance for your response. @Laszlo Bencze

I just upgraded to v1.1.9 as well and the behaviour is still the same

Hi! Is there a way to provide

run_config

for

build_asset_reconciliation_sensor()

somehow?

Hi guys, Is it correct that the gRPC server is only responsible for giving Dagit UI information about the code location? I.e. this code is not serialized and then used by the daemon / k8srunlauncher? Am I correct to understand that the Daemon always fetches the latest image instead of taking what has been serialized from the gRPC server? In our usecase we rsync all dagster assets from a bucket to a dagster repository but if the k8sRunLauncher pulls a new image then that image wont have our dagster assets 😞 Thanks!

Is

SensorEvaluationContext.instance

scoped to the code location it's defined in? Or is it cross-location?

I'm trying to terminate a huge backfill (all runs) but I didn't find any place to do that. What I can perform is select the runs (up to 25) and then terminate. Is it possible to select all of them not only 25? Are there any configuration to allow that?

Hi, can we increase the replica count of the dagit pod to more than 1? Want to make sure that nothing unintended will happen if we increase this CC: @Caio Tavares

Hi Dagster Team,

Hi Dagster! Probably a simple one but what's the correct way to reference a dbt asset inside a

selection

clause for a job?

# dbt_project.yml
models:
  retail_pipeline_dbt:
    stg_ulta:
      +schema: stg_ulta

# dagster_project/assets/dbt.py

dbt_assets = load_assets_from_dbt_project(
    project_dir=DBT_PROJECT_PATH,
    profiles_dir=DBT_PROFILES,
    key_prefix=["ulta_dbt"],
)

# dagster_project/assets/__init__.py

from dagster import define_asset_job
from .dbt import dbt_assets 

asset_job = define_asset_job(name='dbt_job', selection=[?])

Hi, I have a strange issue in dagster and dagster-celery. In my local environment, everything works fine as expected. However, In prod in AWS ECS, sometimes, a task is submitted but i do not see any task in a task queue. in dagit UI event log I can see that the task was submitted to the queue. Then, job get stuck there. Did anyone have the same issue as this? Or if anyone knows about this issue, can you point me how to fix this issue? FYI, dataster = 0.15.2 ann dagster-celery = 0.15.2

Hi, we recently saw this error:

DagsterExecutionInterruptedError

in a few of our jobs … was trying to find what this means in docs but can’t seem to find it. Would anyone be able to help in terms of how to resolve this? I found this on GH: https://github.com/dagster-io/dagster/blob/1.0.17/python_modules/dagster/dagster/_core/execution/plan/utils.py#L84-L94 which if I’m reading this correctly means that this error will get thrown if there’s no retry_policy? CC: @Phil Armour

Is it currently possible to have a multi dimensional dynamic_partitioned_config? The use case here is that we have daily batch jobs that run per customer, but we also want to support back filling by day. New customers are added all the time, which is the need for the dynamic_partitioned_config. Not sure how to get this setup with Dagster and we are currently migrating from Airflow.

hey folks, I am having an issue with the k8s run launcher config that I could use some help with. So the issue is that when Dagster launches jobs, it isn't picking the right node pool. I am adding a runK8sConfig section to my run launcher config but its violating the schema check when I try to helm apply. Helm Error:

runLauncher.config.k8sRunLauncher: Additional property runK8sConfig is not allowed

Config:

"runK8sConfig": {
   "containerConfig": null,
   "jobMetadata": null,
   "jobSpecConfig": null,
   "podSpecConfig": {
      "nodeSelector": {
         "<http://cloud.google.com/gke-nodepool|cloud.google.com/gke-nodepool>": "airflow-ha-pool-2"
      },
      "tolerations": [
         {
            "effect": "NoSchedule",
            "key": "dedicated",
            "operator": "Equal",
            "value": "airflow-ha"
         }
      ]
   },
   "podTemplateSpecMetadata": null
}

which is under

runLauncher

->

config

->

k8sRunLauncher

any advice here?

is there a way to have dagit running in the background as a service instead of in a teminal window?

hey, we've noticed that when our custom SQS sensor yields more

RunRequests

the tick rate for the sensor drops significantly. we were wondering what determines the sensor tick rates?

Hello everyone, does anyone know if we can use the k8sLauncher without the

postgres_password_secret

config option?

Hi, I have a question regarding DT assets which I didn't find in the docs. Assuming that I loaded my DBT assets:

dbt_assets = load_assets_from_dbt_project(
    project_dir=DBT_PROJECT_PATH, profiles_dir=DBT_PROFILES, key_prefix=["jaffle_shop"]
)

How can I access/modify models as individual assets (to add metada, versioning, plots etc.)? mock example:

@asset
def customers_dbt_asset(dbt_asset = customers, key_prefix=["jaffle_shop"], group_name="staging"):
    return customers