Hi All! In my Dagster job I am using dynamic outputs from one Op that get mapped to another Op. I would like to limit the number of mapped Ops executing at the same time. How might I achieve this?

Hi all, I have a question regarding to the asset materialization using DagsterGraphQLClient 1. I have an asset with partitions that defined as:

parts = MonthlyPartitionsDefinition(start_date='2022-01-01')
@asset(
    required_resource_keys={"spark"},
    io_manager_key="io_spark_iceberg_parts",
    partitions_def = parts
)
def parts_proc(context, parts_source):
    parts_proc = some_func(parts_source)
    return parts_proc

2. I have an asset job defined as:

assets_for_job = ['parts_proc']
asset_job = define_asset_job(
    name='asset_job', 
    selection=assets_for_job 
)

3. I have some definition to start job from external jupyter:

from dagster_graphql import DagsterGraphQLClient
client = DagsterGraphQLClient("dagster", port_number=3070)

client.submit_job_execution(
    job_name="asset_job",
    repository_name="my_repo"
)

How could I set the list of asset’s partitions that I need to materialize in “submit_job_execution” at the step 3?

Hi guys, part of Zephyr AI here, we have been using dagster for a long time and have routine philosophical conversations like “what if we used dagster to run ci/cd for us”. Jokes aside, i have a support question on dagter 1.1.17. We have an issue where: we have a long running pipeline (lets say 18 hours) on k8s where we use the pod per op (pod per step) strategy. our hypothesis is when: the dagster daemon restarts sometime during the course of the execution of the pipeline (so any one of the steps) then: the next step in the pipeline is not spun up the dagster run pod and dagit just show the job in progress, thats it. this is a rough timeline: • run kicks off at 17:54 • 6 minutes later (18:00) a step pod kicks off • during execution (03:22) daemon restarts •

Ignoring a duplicate run that was started from somewhere other than the run monitor daemon

gets logged in dagit at 03:23 • step pod runs for 11 hrs and terminates at 05:42 • next step pod is not spun up ive been told dagster support is pretty responsive, and this is becoming quite a headache for us, so any pointers would be much appreciated.

Having a beginner level question, not sure if it's a bug (very likely I'm doing something wrong): I'm 1. definging a SourceAsset with metadata like pic 1; 2. try to use the metadata in my custom io manager like pic 2; 3. then I expect to see the defined metadata using

context.metadata

. from the debug info in pic 3 you can see: the asset key is present, but there is no metadata (which shoud exist I think?) for github issues, I only searched and found https://github.com/dagster-io/dagster/pull/6900, so I think my usage is valid, is that true? Thanks!

pictures here.. sorry not familiar with slack

This is a cross-post from the #dagster-snowflake channel, per recommendation of Jaime... I have a question that maybe related to how I'm using dagster with Snowflake or might be that I'm not storing assets properly, etc. This is with a multi-container docker setup. I have a module that I'm mounting as a volume, per the docs. The structure is similar to this example, where I have packages/folders in my assets directory. In the linked example, there are packages for

activity_analytics

,

core

, and

recommender

. In my case, let's say I have

package1

and

package2

. As with the example,

all_assets

ends up being

[*package1_assets, *package2_assets]

and is passed to

Definitions

in the same way that the assets are in the linked example. If I go into dagit and launch a run to materialize all of the

package1

and

package2

assets, which are in different groups, all works fine. However, if I modify the code to

package2

and then update the git repository and restart dagster/dagit, the system no longer knows that the

package1

assets were materialized, in spite of the fact that

package1

did not change at all. I can understand that

package2

would reset, however, I'd like to figure out how to set this up so that only the assets associated with the package that changed would need to be rematerialized for dagster to know about them. Most of my work/assets is being pushed to Snowflake. Is this simply a matter of dagster not knowing how to find them on Snowflake? Or will all of the assets associated with all of the packages always reset when I push a code change to a single package?

Man I know this should be simple, but I'm having trouble following the documentation. Could someone help me further understand how I'm supposed to configure aws secretsmanager as a resource if I'm using a Definitions object on init? Apologies in advance if I'm being dense.

import dagster

from dags import cf, observations, ts, pr
from models import database

defs = dagster.Definitions(
    jobs=[
        observations.import_observations.to_job(resource_defs={"database": database}),
        cf.copy.to_job(),
        ts.ts.to_job(resource_defs={"database": database}),
        pr.ss.to_job(resource_defs={"database": database})
    ],
    resources={"database": database},
)

Is it possible for a partitioned asset to depend upon a non-partitioned asset?

Hey Dagster community: I have an op that schedules a k8s job using

execute_k8s_job

. The op is called via

map

of a `DynamicOutput`list.

execute_k8s_job

uses the name from the

OpExecutionContext

as the k8s job name. However, it doesn’t appear that the op’s name is actually unique for each invocation of the op in the map operation, which results in dagster attempting to create k8s jobs with duplicate names. I’ve considered using a factory that generates ops with unique names, but am unsure how I’d use this with

map

. How do I ensure that each mapped invocation of an op has a unique name?

Hi guys, we are trying to launch dagster on EC2, and got below errors, when try our first test run. Can you list some possible causes? Thx

Hi. Is there a way to pause a job (say because an external device needs input) and then unpause the job so it can continue? The hope would be to provide a signal so that the user at the web UI knows.

We have jobs that run on a schedule every couple of hours. And when they fail they are setup to send a notification to on call people. The issue is fixing the problem might take more than a couple of hours, especially if a failure happens in the middle of the night. This can result in a piling up of notifications that are redundant and noisy. Airflow has

depends_on_past

so you can prevent a job schedule from executing if the last run of the same job failed. Is there an analog to this for Dagster? I'm seeing execution_fn and should_execute that leverage

ScheduleEvaluationContext

but that context seems to only have information about the current run. Any best practice suggestions for this situation?

Hi Team, facing below error during helm re-installation of the dagster. Any help on this issue?

FATAL: "/bitnami/postgresql/data" is not a valid data directory
DETAIL: File "/bitnami/postgresql/data/PG_VERSION" does not contain valid data.
HINT: You might need to initdb.
chmod: changing permissions of '/var/run/postgresql': Operation not permitted

Hello dagster community! I am excited to be here! I have a question about the design of my workflow I am designing a workflow that includes some heavy compute tasks (mostly ffmpeg tasks). if I understand correctly, it is a bad practice to run compute heavy tasks in the context of the dagster executors. so I thought about extracting these workloads to an external service and make dagster just send requests to these service. my questions are whether this is a good approach or not, and if it is, who should handle the materialization of the files after processing? my external service can upload the processed files to an object storage and return the path to dagster, or it can return the whole huge byte array over the network and let dagster persists that to an object storage (that sounds bad to me). how would you handle this pipeline? thanks!

Hello! Can someone explain briefly if is possible to load data resulting from an extraction into a Postgres database? It is through IO Managers or something like that? Thank you!

I have a custom DataFrame type, like:

MyDataFrame = create_dagster_pandas_dataframe_type(...)

And an asset that produces that type, for storage in Snowflake. This all works as expected. The asset signature is like:

@asset
def my_asset(context: OpExecutionContext) -> MyDataFrame:
    ...
    return df

Validation passes, etc. Downstream, I have an asset that needs to understand the types specified for columns in

MyDataFrame

. I tried to do this for the downstream asset's signature:

@asset
def my_downstream_asset(context: OpExecutionContext, my_asset: MyDataFrame) -> None:
    ...
    # some stuff happens
    ...
    return None

However, I cannot materialize

my_downstream_asset

because of the error:

dagster._check.CheckError: SnowflakeIOManager does not have a handler for type 'typing.Any'. Has handlers for types '<class 'pandas.core.frame.DataFrame'>'. Please add <class 'pandas.core.frame.DataFrame'> type hints to your assets and ops.

I had this same error yesterday when I did not specify a type at all for

my_asset

in the signature of

my_downstream_asset

. How can I have

my_downstream_asset

load

my_asset

and apply the column types specified (and validated) in

MyDataFrame

?

Is the

SnowflakeIOManager

ignoring the

MyDataFrame

type hint?

or does this need to be specified in

@asset()

somehow? I tried adding the following to

@asset

and had the same error:

ins={"my_asset": AssetIn("my_asset", dagster_type=MyDataFrame)}

The only way I'm able to load

my_asset

in

my_downstream_asset

is with the signature:

def my_downstream_asset(context: OpExecutionContext, my_asset: pd.DataFrame) -> None:
    ....

... or, is the metadata about the MyDataFrame type stored in the dagster postgres db so that the types automatically are applied? 🤔

Hopefully a quick one. If we have a bunch of assets, say 60, and these materialize in parallel (8 at a time) but one seems to take way longer than the most, we would like to suggest to/force dagster to start materializing that asset in the first batch of assets that get materialized. Is this possible (without artificially constructing our graph with this asset being an input into all other assets)?

Is there a way to get dagster to materialize everything from the command line, or the module in python? I want to set up integration testing and one would simply be to run through every single materialization (like I went to the web UI and clicked "materalize all" on every group) and make sure nothing fails

I'm trying to figure out the template provided GitHub action workflow

branch_developments.yml

process. I'm running Cloud/Hybrid. • What is my

DAGSTER_CLOUD_URL

• The

DAGSTER_CLOUD_API_TOKEN

is a "user" token from the GUI? • My

ORGANIZATION_ID

is

klar

I figured that out (ha!)

Hello! 👋 When unit testing asset pipelines in a CI pipeline (e.g. GitHub/GitLab) is it possible to validate the asset pipelines' execution plan without materialising the asset, like a job with the

dagster.core.execution.api.create_execution_plan

function? Reason being some assets i.e. data from APIs, can't be materialised outside production.

Hello everyone, I am getting this error message, not sure what it means and why it happened. Dagster version: 0.12.15

Hi All, Just flagging this error that I ran into during local testing:

Sorry, page can't be displayed.
Please report this error to the Dagster team via GitHub or Slack. Refresh the page to try again.
Loading CSS chunk 710 failed.
(<http://localhost:3001/static/css/710.59a51a08.chunk.css>)

Error: Loading CSS chunk 710 failed.
(<http://localhost:3001/static/css/710.59a51a08.chunk.css>)
    at o.onerror.o.onload (<http://localhost:3001/static/js/main.9a8fb596.js:2:1818747>)

Hi everyone, I have a quick question. Can dagster work with Flink? Thank you@@

Is there a way to mark partitions as Skipped or Succeeded? • The use case here is that I want to re-partition an asset, but don’t want to have to backfill. I also don’t want partitions that previously ran to be considered “missing” • ex. Imagine I have partitions defining ranges 0-99, 100-199 …, but want to change the partitions to represent 0-9, 10-19 … Related, is there a max # of partitions that an asset could have (hopefully without causing perf regressions in Dagit)

is there documentation somewhere for how to use

context.asset_partition_key_range_for_output

/

context.asset_partition_time_window_for_output

to enable a partitioned asset to operate on a range of partitions? I'm currently using

context.partition_time_window

and

context.asset_partitions_time_window

and am a bit unsure of how to modify my code to use this feature

Hi, hopefully someone has seen this before. Is this a code or infrastructure error?

TypeError: SqlEventLogStorage._get_asset_entry_values() missing 1 required positional argument: 'has_asset_key_index_cols'

What happens to io at an op-graph boundary? If the last op in a graph writes to its output iomanager, does the graph-job shell re-pickup the output as input and store it elsewhere?

By implementing a custom io-manager i am able to trace two writes: so am guessing one at op end (its the last op in the graph) and one at graph/job end. How do i connect last-op outputs to the graph to ensure this does not happen?