Hi there, We have been using Dagster for a few months now and rely of Sensors quite heavily. Unfortunately, we are now facing issues where the Sensor page in Dagit is not loading, therefore, we are unable to enable/disable sensors. The version of Dagster that we are using is 1.0.9. Can anyone please let us know how we can clear/fix the issue when loading a list of Sensors in Dagit UI? Thanks in advance

Hey everyone, I’m trying to create a source asset which represents a table in my db that is managed completely outside of Dagster (it’s essentially a manually curated set of ids that we want to rely on in the pipelines with sensors). I’d like to create an

observable_source_asset

for this, but I need to have access to a resource (db connection) within the

observe_fn

to calculate a

LogicalVersion

value. For a normal asset I have

required_resource_keys

and the

context

object, but how do I do this in an observable source asset? Does this even make sense?

Hello - I need to load information into a DB from x number of upstream assets. I thought I might be able to do this by using a multi-asset and passing some recursive function to dependencies that materialises the downstream assets once the parents are reconciled, but I can't seem to figure out how to do that without repeating logic. In other words, I need to create x number of assets given x number of dependencies. Anyone tried something like this before?

Hello, is there any specfic reason we can’t set

retry_policy

on a

@graph

and

define_asset_job

? As a workaround we can set the retry policy on the assets and ops but since this property exists for

@job

it would be nice to have it.

Hello! I'm testing out the new definitions way of loading assets/jobs (instead of repositories) in dagster 1.17. However, the jobs don't seem to be loading in Dagit. The assets are loading in dagit, but the job is not loading (on the side bar there would be a "job" tab, also there are no job definitions loaded when I run dagit)

@asset

def asset1():

return [1, 2, 3]

@asset

def asset2(asset1):

return asset1 + [4]

all_assets_job = define_asset_job(name="all_assets_job")

asset1_job = define_asset_job(name="asset1_job", selection="asset1")

defs = Definitions(

assets=[asset1, asset2],

jobs=[all_assets_job, asset1_job]

I stole the code snippet from the "how to create jobs from software defined assets section of the docs" https://docs.dagster.io/concepts/ops-jobs-graphs/jobs#from-software-defined-assets Am I doing something incorrect with the new API?

message has been deleted

Hello all. We have a Dagster setup that uses Poetry to manage the virtual environment. This is our `workspace.yaml`:

load_from:
  - python_package:
      package_name: pipeline
      executable_path: .venv/bin/python

Everything works great on Linux, but Windows uses a different path to the Python binary:

load_from:
  - python_package:
      package_name: pipeline
      executable_path: .venv/Scripts/python

Maybe this is more of a Poetry question..

Hi there, How can I set an Asset depending on an specific MultiPartitionKey? My example: I have a multi-partitioned asset and want dependency only in one partition, something like:

AssetIn(key=, partition_key=MultiPartitionKey({..})) # pseudo-code

@sean --> any pointers on how to possibly solve this 👇 .. up until yesterday our dagster job was running normally and yet today it has stopped working.

Details --> job is running on local machine, dagit==1.0.17, dagster==1.0.17

I’m working on a slack run failure sensor. Unfortunately, there is a backlog of failed runs it is going through from the past few weeks. It would be nice for the error message to show the date the job kicked off or failed - is this data available in the RunFailureSensorContext?

i'm trying to follow the docs for creating a partition asset job (and eventually want to schedule partitioned assets to materialize daily). My asset works and I can manually materialize and backfill assets, but when I try to add a job (so that I can eventually add a schedule) the code deployment fails `dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code serv`er. I keep going over the docs and am not sure where I am going wrong. If someone could point me to a good example or help me 1. understand if this is the correct way to add schedules for materializing partitioned assets daily - the plan is to get the job working and schedule the job using a scheduler 2. help me understand why I am unable to load this job. Please see code below

Hey there! We are trying to figure out what the best way of viewing some of our notebook outputs would be. Currently after the op is run, the notebook is saved to azure (we use papermill not dagstermill since these are spark jobs). It would be neat if we could somehow output the notebook as an event and have it render in dagit. Do you have any thoughts as to how I could start implementing something like this? EDIT: I see that there is already some way to view the notebook output when using dagstermill. However since we are running this job outside of dagster on spark cluster we use papermill directly. Is there a way to get the same UI treatment?

Hi Team, We are deploying dasgter on EC2. We have two jobs as of now which run hourly for 5mins each. Any idea on what AWS instance size would be appropriate for this deployment ? Are there any minimum specs recommended.

How would I pass a config option for an io_manager from an asset? Let's say I have a redshift io_manager that per default appends to tables but I have certain assets that need to do a replace instead. How would I go about this?

hi, how can I set

op_retry_policy

on

define_asset_job

?

tags={"dagster/max_retries": 2}

would probably work. But I'm also interested in controlling delay between retries..

Hi everyone, I am trying to create a graph dynamically, meaning that the number of Ops in a graph is determined by an external configuraiton. The use-case is that I want to run spark jobs in steps, and one job might have 5 transformations while another might have 3. The Op launches a spark job on EMR and is templated. The output of one op is not the input of the other. I simply want them chained in a certain order and I don't want to have to hard code the Ops by using the

@graph

decorator on a function and specifying the Ops inside that function. I am getting the following error due to the

Op

which has a

hook

. The error:

AttributeError: 'PendingNodeInvocation' object has no attribute '__name__'

. This happens because

GraphDefinition

expects

node_defs

to be a list of

NodeDefinition

and not

PendingNodeInvocation

. I tried to separate the hooks from the nodes and pass them to the repository but a repository does not accept the

HookDefinition

class. I want to apply certain hooks to specific Ops so applying a hook to the entire job doesn't fit my use case. Is there a way I can work around this? or a different way I can create Ops dynamically in the same way I want to? Below is an example of how I create my code:

from dagster import GraphDefinition, op, DependencyDefinition, OpExecutionContext

nodes = list()

def get_template_node(x: int):

@op(name=f"node{x}")

def template_node(context: OpExecutionContext, some_value) -> str:

return f"This is node {x}"

return template_node

def get_downstream_node(x: int):

@op(name=f"downstream_of_node{x}")

def downstream_node(context: OpExecutionContext, some_value) -> str:

return f"DOWNSTREAM GOT VALUE {some_value}"

return downstream_node

for x in range(0, 5):

nodes.append(get_template_node(x=x))

nodes.append(get_downstream_node(x=x).with_hooks({my_hook_definition}))

dependencies = dict()

for x in range(len(nodes) - 1, -1, -1):

if x != 0:

dependencies[nodes[x].name] = {

"some_value": DependencyDefinition(nodes[x - 1].name)

}

my_graph = GraphDefinition(

name="dynamic_graph",

description="dynamic grap desc",

node_defs=nodes,

dependencies=dependencies,

)

my_job = <http://my_graph.to|my_graph.to>_job(name="dynamic_job")

@repository

def repo():

return [my_job]

Hi all, Is it possible to create an asset job that includes assets from another repo? We want to create workflows that sequentially materialize (dependent) assets across repos. E.g.:

my_workflow_job = define_asset_job(
    "my_workflow_job",
    selection=AssetSelection.keys(asset_from_upstream_repo_A, asset_from_upstream_repo_B, asset_from_this_repo),
)

If this is not possible, what’s the best way to create such workflows?

We use op_config in ops then how can we pass op_config or context.op_config in a graph .. how can we pass it as parameter and use it..? Can anyone pls help with syntax example..

When starting a job I am getting

dagster._core.errors.DagsterLaunchFailedError: Error during RPC setup for executing run: dagster_postgres.utils.DagsterPosgresException: too many retries for DB connection

but other jobs work just fine. Any ideas what could be happening?

Should the config provided in Launchpad override

.configured

config? My resource:

@resource(
    config_schema={
        "host": Field(StringSource, default_value="localhost"),
        "db": int,
        "password": Field(StringSource, is_required=False),
    }
)
def redis_client(init_context: InitResourceContext) -> Redis:
    ...

The Launchpad config:

resources:
  redis_client:
    config:
      db: 1

I pass a "partially " configured (I realize it's actually fully configured) resource to my `Definitions`:

"redis_client": redis_client.configured({"host": {"env": "REDIS_HOST"}, "db": 0}),

After this the

db

value is always 0, while I would expect the Launchpad config to override the value provided from the code.

Hi! I have the current situation: • A partitioned asset A • A partitioned asset B which depends on asset A When my io manager tries to get data from A (when trying to materialize B), it is failing because it’s looking for a non-partitioned asset, i.e.

context.upstream_output.has_partition_key

is

False

. Why could this be happening? Both assets are annotated with

@asset(partitions_def=DailyPartitionsDefinition(start_date="2015-02-10"))

and their job definitions include

partitions_def=DailyPartitionsDefinition(start_date="2015-02-10")

as a parameter. I need the io manager the pick up only the partition from A that corresponds to the same date of B that I’m trying to materialize.

With 1.1.6 or 1.1.7 (not sure about earlier versions), when I run

dagit

I get

cannot import name 'introspection_query' from 'graphql'

. Anyone else seeing this? I tried reinstalling my requirements in a fresh venv. And I think my only graphql requirements are coming from Dagster. I can call

execute_job

or run via Dagster Cloud just fine.

Greetings. I'm encountering a configuration error when loading my definitions: I have a

job

that calls an

asset

that itself has two upstream assets. The

job

has a

config=

set for one of the

ops

(via

config={"ops":{"op1": ...}}

). When I attempt to load the repository, I get an error like so:

Missing required config entries "asset1" at the root. Sample config for missing entries: {'asset1': {'inputs': {'upstream_asset1': '...', 'upstream_asset2': '...'}}}

If I add the above scaffold to my `job`'s

config=

stanza, I get another error:

Error 1: Must specify a field at path root:asset1:inputs:upstream_asset1 if more than one field is defined. Defined fields: ['json', 'pickle', 'value']

I'm not sure why this configuration is needed (I did not define any

config_schema

for any of the assets), and I also wouldn't know what are the correct values to place in these config sections. I can't find anything in the documentation (or by reading the dagster source code, even) that helps me figure it out. Thanks for any help/pointers you can offer!

Is there a way to define a job to materialize a selection of assets for a specific set of partitions? e.g.

@asset(partitions_def=StaticPartitionsDefinition('abc'))
def partition_samplestep(partition_sample) -> pd.DataFrame:
   ...
   return partition_sample
part_job = define_asset_job(name='partition_job', selection=partition_samplestep, partitions_def=<some syntax to select only partitoin c>)

Does calling an

@asset

function from a

@job

cause the asset to be re-materialized, or does it use whatever was the last materialization value?

Hi, I am getting following error in Dagster UI. Can anyone suggest how to resolve it? Operation name: AssetGraphLiveQuery Message: maximum recursion depth exceeded while calling a Python object Path: [“assetNodes”,16,“projectedLogicalVersion”] Locations: [{“line”44,“column”3}]

Is there a way to reuse the output of an op if it has already been calculated?

Hello! Are the plans to add support for Python 3.11? Thanks!

Hello!. Is there a straight-forward way to generate a non-partioned asset from a partitioned upstream asset. In my case, I have 50 csv files each representing a US state, and I wish to load them into one table. It feels sensible to define the group of files as a partioned asset but I couldn't quite work out how to then generate the downstream asset - is the set of partitioned files available as an iterable in the op?

I have a question. what is the reason not to have multiple OPs with the same name although these OPs belong to different Jobs in a Repository?