Hello Dagster, you might be interested to know that there are still a few "solids" and "pipelines" references in the latest documentation, and it seems you might have forgotten to reword them: • https://docs.dagster.io/_apidocs/libraries/dagster-celery#api • https://docs.dagster.io/_apidocs/libraries/dagster-dbt • https://docs.dagster.io/_apidocs/libraries/dagster-airflow • https://docs.dagster.io/_apidocs/libraries/dagster-msteams • https://docs.dagster.io/_apidocs/libraries/dagster-mlflow • https://docs.dagster.io/concepts/ops-jobs-graphs/op-retries#retrypolicy

This page can't seem to load and returns "Application error: a client-side exception has occurred (see the browser console for more information)." • https://docs.dagster.io/_apidocs/libraries/dagster-pandera

Hi folks - is there a way to utilize a cursor in a

RunStatusSensor

? I couldn't find a way looking at the source or docs, but do see there's a "cursor" field in the UI

Hello D team, I am using Dagster with DBT and I am struggling with the partitioned assets. Ideally I would like to have a job that will materialize last n (refresh window) partitions and then lunch a DBT step for these partitions. I want to make sure that DBT steps are not running in parallel for each partition, that's why I would like to have it at the end. The only way I could think this can be achieved is by using AssetMaterializationEvents or by using sensors, but neither solutions seam robust enough. Am I missing something? Thanks, Slaven

Hi, is there a way to have a default input for graph? something like:

@graph
def calculate_distance(metric: Metrics = Metrics.kilometer)
    pass

such as

Metrics

is an enum? I’m getting:

input metric must get a value from the inputs section of its configuration.

Can you configure a RetryPolicy for a

k8s_job_op

? I declare my op like this:

def function_to_create_k8s_op( <a lot of variables > ):

    op = k8s_job_op.configured(
        config_or_config_fn = {
            'image': f"{image_url}:{tag}",
            'service_account_name': env['SERVICE_ACCOUNT'],
            'command': ['java'],  
            'args': args,
            'resources': {
                'requests': {
                    'cpu': cpu,
                    'memory': memory,
                },
            },
            'image_pull_policy': 'Always',
            'env_vars': all_env_vars
        },
        name = name,
    )

    return op

I tried

op.retry_policy = RetryPolicy(...

but that attribute is read-only. My understanding is that

k8s_job_ops.configured(config_schema

should only be used to set the config schema of the op, which is already defined here. I have also tried setting other attributes by passing them into

job_spec_config

, but I found that those were being ignored, and the only way to get attributes set was to pass them into

config_or_config_fn

(which seems like a a bug, but because I'm using an older version - 1.0.2 - i'm not focusing on that). So is there any opening to insert a retry policy? Or would the best option be trying to upgrade to the newest version and then using

job_spec_config

? If so, I also could use a pointer to the correct K8s spec argument for retries, because in my last run through the docs I didn't see any.

Hi dagster team! I'm catching strange behavior on our dagster instance. We deployed dagster on kubernetes. We run a lot of dbt models in our dagster ops. The problem is that sometimes jobs are freezing on dbt ops. It look like pod has restarted and provide database connection lost, but dagster doesn't handle it correctly and still waiting for response from database. Is there some way to deal with it? Thank you!

Hi all! 😄 What's the most straightforward way to pipe logs to Datadog?

Thread: Does

snowflake_io_manager

support non-date partitions?

Hi ! I am facing issue with long running meltano job in dagster. After almost 6 + hours of execution of job I am getting below errors: MYSQL Connection lost

"run_id": "f72306b3-xxx-xxxxx", "state_id": "el-data", "stdio": "stderr", "cmd_type": "extractor", "name": "eldata", "event": "pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')", "level": "info",

Some time below pod restart error:

started a new run worker while the run was already in state DagsterRunStatus.STARTED. This most frequently happens when the run worker unexpectedly stops and is restarted by the cluster. Marking the run as failed.

Please let me know any one had faced these issues and how it was resolved. 🙏

Is anyone else battling with the free version of slack now to search for help? We can now only search 90 days back for previously asked questions, right? I wonder what the decision here was? As Slack is used much more than StackOverflow, I wonder how sustainable this is, and perhaps it's good to also start posting more questions on SO where they will be able to be found by others with the same question?

Hi, Is there a way to select a subset of asset keys based on their hierarch, eg. every asset key that starts with

foo/bar

(but possibly contain more levels, eg.

foo/bar/baz

is matched as well as

foo/bar

and

foo/bar/baz/*qux*

). I’m trying to avoid explicitly listing all the assets in a

define_asset_job()

.

Hey, is it possible to pass name to

k8s_job_op

without using

configured

method? I would like to pass different configs for two k8s_job_ops in job config at runtime. It seems that I need to specify names. Or is there another way to pass config to different

k8s_job_ops

at runtime? i.e. something like this:

@job
def my_job():
    k8s_job_op(name="add")
    k8s_job_op(name="substract"))

my_job.execute_in_process(run_config={"ops": {"k8s_job_op": {"config": {"args": ["the", "args"]}}}})

The problem is that I need several

k8s_job_op

. It works with one sadpanda

Hey, we’d like to create a sensor that reloads the workspace upon some event (we have dynamic repositories to generate jobs). What would be the best way to invoke the reload operation through the daemon? I know it can be done using

DagsterGraphQLClient.reload_repository_location

but I have to get somehow Dagit’s IP and all the repository locations. Is there a way to extract these from the sensor context? (maybe from the

DagsterInstance

)? If there’s another way (not using GraphQL and Dagit) that could be great too! Thanks!

Hi All, We are trying to use @failure_hook (applied on job level) to get failure messages for the ops to go to our slack channel. We have added this failure hook to a job that has multiple ops in it. One of the ops failed with this following message but it didn't invoke the failure hook. Can you please help us understand why the hook wasn't even called? Note: For other ops failure, we saw hook invokations in the dagster logging console after the op was executed. But for this particular op, there were no log events showing hook was called or skipped and the job went on to execute the next op. - Eg :

Skipped the execution of hook "slack_message_on_failure". It did not meet its triggering condition during the execution of "xxx".

Ola! Does anyone know if there's a way to get an overview of all failed (and maybe also missing) partitions for all Assets in one place? I've not been able to find one.

Hey there! I have a job where I have configured the activeDeadlineSeconds to be 600. However the job has been running for 10 hours. When I describe the job in k8s I see: ''' Active Deadline Seconds: 600s Pods Statuses: 0 Active / 0 Succeeded / 1 Failed "'' So it seems the pod failed but the job is still active in dagster

Does anyone have insight into why I might be getting a

UserWarning: Error loading repository location $dbt_project:KeyError: 'model.$subfolder.$model_name'

Error when trying to load in a DBT project? I've followed the basic folder structure from the github DBT/Dagster example, with both my DBT project and Dagster project at the root of the PWD, with the same profile/project import structure as in the example

Is there a way that I can tell Dagit that I already have partitions of an asset? E.g., when creating a branch deployment, we create a branch database and copy some raw data over so that we don't have to sync it again. I'd like Dagit to be aware of these partitions that are already present.

Hi all, I have a partitioned asset which I need to run scheduled materializations every 5 minutes (there's an application accessing its data and reporting it with 5-min update frequency). The materialization consists on extracting the data from an external API and writing it to my database. However, this process can have some random failures along the day, so I end up missing a couple of 5-minute partitions, and the are usually clustered togheter. For example, I can be missing, say, partitions from 13:00 to 14:20 and also from 18:50 to 19:30. I can use the Materialization UI to re-launch those materializations, but it will queue one execution for every 5-minute partitions, which would be inefficient. The API supports calls with arbitrary time ranges (up to a maximum span), so it would be great if I could tell dagster to launch a custom Job to query the API for the time range to save the data to the database and yield corresponding

AssetMaterialization

to maintain asset materialization consistency on dagster when a materialization of multiple partitions of a partitioned asset is required. I know it is possible to just launch a Job to do exactly that, but imho it defeats the great usage flow which comes from exploring the asset list and re-materializing a couple of partitions.

Hello! I’ve created an IO Manager that reads and writes pandas dataframes to a sqlite database. We use sqlite because our data is fairly small and we want to enforce primary and foreign key constraints. IO Managers that read and write to databases seem like an accepted dagster pattern (see snowflake example and ). There are two downsides to a sqlite database IO Manager. First is datatypes often get screwed up when moving between pandas and a database. Second is writing and reading to and from a sqlite database is much slower than writing to pkl or parquet file. Do people run into these two issue when working with the Snowflake IO manager? Is this a reasonable pattern to be using? Thanks!

Detecting if a file has finished downloading I use dagster jobs to download files using a web browser (using

selenium

and headless firefox+geckodriver) and then read those file into pandas DataFrames and then load the data into a database (or whatever). I have written some polling functionality (using

polling2

) which helps me to work out once a file has fully downloaded, but I'm not convinced that it is always accurate. Sometimes the files are largish (~200MB). There is a significant lag between the downloaded file initially being created in the filesystem and the all the data being written into that file. I first poll for the existence of the file and then I poll for that file having non-zero size. Is there a better way (on Linux) to confirm that a file is no longer being written to and is therefore ready to be consumed?

dagster-k8s not setting ttlSecondsAfterFinished I use the K8sRunLauncher to run jobs (dagster 1.0.3, kubernetes 1.20.6), but it does not set

.spec.ttlSecondsAfterFinished

in the Job resources it creates resulting in a lot of accumulated Jobs never being deleted. I tried to set the

dagster-k8s/config

tag explicitly like described in the documentation but to no avail. I also tried updating to 1.0.17 but that didn’t help either. Is there something else I need to consider here?

Hi dagster team! I have a question about K8sRunLauncher. How can i specify ram/cpu/gpu for particular deployment/job with this launcher?

Hi dagster Team, I am studying dagster graph nowadays and I’d like to ask about your example. I can see the graph with dagit -f filename.py. And I have found the storage folder but I can’t find in the result files the result. I don’t know where is the final result of the sum_fan_in(fan_outs).

from typing import List

from dagster import graph, op


@op
def return_one() -> int:
    return 1

@op
def sum_fan_in(nums: List[int]) -> int:
    return sum(nums)


@graph
def fan_in():
    fan_outs = []
    for i in range(0, 10):
        fan_outs.append(return_one.alias(f'return_one_{i}')())
    sum_fan_in(fan_outs)

What is the best way to process a very large dataset by splitting them to smaller chunks ? I have dataset coming in for different customers and I wanted to process each customer separately. The data is first loaded from the DB and each subset of a customer’s data is large enough that keeps going on for 10s of hours I want to be able to process each customer’s data separately and in turn it will split each customer’s data by date and process per date chunk separately Its more of a multi level fan out structure

Does anyone know of an example using

@multi_asset

that also has a downstream dependency asset?

Hello lovely support. I think I may have found a bug. I’m trying to create an Asset from a graph. We use the

gcs_pickle_io_manager

IO Manager. Here’s a close version of my code:

import os

from dagster import op, graph, AssetsDefinition, ResourceDefinition
from dagster_gcp import bigquery_resource, gcs_resource
from dagster_gcp.gcs.io_manager import gcs_pickle_io_manager

@op
def upstream_op():
    return 1


@op
def downstream_op(upstream_op):
    return upstream_op + 1


@graph
def my_graph():
    return downstream_op(upstream_op())


asset_definition = AssetsDefinition.from_graph(
    my_graph,
    group_name="my_group",
    resource_defs={
        "gcs_bucket": ResourceDefinition.hardcoded_resource(os.getenv("BUCKET", "bucket")),
        "gcs": gcs_resource,
        "bigquery": bigquery_resource,
        "io_manager": gcs_pickle_io_manager,
    },
)

The error I’m getting:

Conflicting versions of resource with key 'io_manager' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.

I’m not sure I understand what this error means. I tried a couple different things but I was not able to fix it 😞 Any help is greatly appreciated

Hello all! I might have found a problem in the documentation: the Cloud getting started guide talks about a "Add sample code location" button (perhaps called "Add code location", per the screenshot there?), but that's not what I see when I look at my Workspace:

The code location docs state "Dagster Cloud doesn't need access to your image", but I can't proceed through the Getting Started guide until Cloud can access my image: