Hi everyone! How can I wait for the completion of the previous

RunRequest(...)

when using a sensor?

Hey team, do you have any docs about Dagster scalability? E.g. how many dynamic partitions can it handle (order of magnitude)? How many runs can it do per minute? What are our options to scale it? Both Cloud and OSS options are interesting.

Hey there, I feel like I might be missing something obvious, but is there something extra I have to do to get `asset`s to use cached results upon re-execution (w/o any changes to their input) 🧵

with pythonic resource definitions is it still possible for resources to access contextual information like the

run_id

during initialization?

Is it possible to attach metadata with

Output

to individual

AssetOut()

in a

multi_asset

?

so I've been happily plugging away with a single code location (python module and definition, excuse the incorrect terms, still getting use to the dagster lingo) connecting to a single instance of our API. I've now been asked to connect to our pre production instance as well. Am I on the right track with copying my

__init.py

from the module to a new file, replace server names in the resource configs, add python_file to the workspace.yaml. (here is where I'm unsure) I'll need to change/add key_prefixes and group_names right?

Any ideas why this sensor is throwing this error

dagster._check.CheckError: Expected non-None value: None
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/errors.py", line 206, in user_code_error_boundary
    yield
  File "/usr/local/lib/python3.9/site-packages/dagster/_grpc/impl.py", line 328, in get_external_sensor_execution
    return sensor_def.evaluate_tick(sensor_context)
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 428, in evaluate_tick
    result = list(self._evaluation_fn(context))
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 598, in _wrapped_fn
    for item in result:
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/run_status_sensor_definition.py", line 589, in _wrapped_fn
    external_repository_origin = check.not_none(
  File "/usr/local/lib/python3.9/site-packages/dagster/_check/__init__.py", line 1081, in not_none
    raise CheckError(f"Expected non-None value: {additional_message}")

I'm trying to design an asset that is the inference results of an ML model for "events". Each event produces one inference result. Events have string ids. We'd also like to experiment with different ML model versions, compare them, so we need the inference results for the same event across multiple versions, built on demand. What would you recommend to represent this structure in dagster? Is it 2-d partitioning? Can we use configuration to distinguish between model versions (I assume we can't as they will overwrite each other)?

Hi folks, is it possible to setup airbyte assets as an upstream of another asset?

Hi team, We have the following file (referred as services.py) that we want to load as a code location. The dbt compile takes a while to run, which is why we have placed it inside a job and placed a schedule on it. We have another code location that loads the dbt assets from the manifests generated by this job. Two questions: 1. When running dagster dev -f services.py, this job seems to start executing and the code location does not load due to the error 'Exception: Timed out waiting for gRPC server to start after 180s'. 2. When viewing the logs, why do we see the dbt compile issue twice for each customer? Does building a dbt resource issue a compile command?

from dagster import Definitions, build_resources, job, ScheduleDefinition
 
 import datasource_lake_defs 
 
 from custom_modules import spina_classes
 from custom_modules import helper_functions
 from custom_modules import dbt_functions
 from custom_modules import armada_functions 
 
 customer_code_list = []
 
 @job
 def job_generate_customer_dbt_manifests():
     # get the org list from armada
     organisations = armada_functions.get_organisations()
 
     organisation: spina_classes.Organisation
     for organisation in organisations:
         # get the config from armada]    
 
         run_config = armada_functions.get_config_from_armada(organisation)
 
         if run_config.datasets:
             customer_code = helper_functions.clean_object_name(organisation.organisation_name.lower())
             customer_code_list.append(customer_code)
             customer_name = organisation.organisation_name
 
             with build_resources(
                 resources={"dbt_res": datasource_lake_defs.configured_dbt_cli_resource},
                 resource_config = {
                     "dbt_res": {
                         "config": {
                             "vars": dbt_functions.get_dbt_vars_for_customer(customer_code),
                             "target_path": f"target/{customer_code}"
                         }
                     }   
                 }
             ) as resources:
                 resources.dbt_res.compile()
 
 schedule_job_generate_customer_dbt_manifests = ScheduleDefinition(job=job_generate_customer_dbt_manifests, cron_schedule="*/10 * * * *")
 
 
 defs = Definitions(
     jobs=[job_generate_customer_dbt_manifests],
     schedules=[schedule_job_generate_customer_dbt_manifests])

@Manan P

Hi, I have a question regarding partitions - I've hit a bit of a wall either will my understanding or not sure how to achieve what I'm trying. • I have a partition asset with two dimension - the date and a static array of parameters (time window length, but not sure it matters for this). I have an IO manager used for the in on an asset that reads values from a DB using the date and window length as parameters. The asset then calculates some statistics from values in those data frames. • My current mental model is an asset generated for every pair of (date, window length) • The upstream data the IO manager reads from isn't partitioned and is a dbt created asset. • I have a job that runs at 5am based on a cron that materializes the dbt asset. Now I'd like it to also materialize the downstream partitioned assets for all time windows and the current date. I guess my question is really am I viewing this correctly in terms of the partitioned model, and how do I achieve my goal of the cron job materializing the dbt asset daily which then materializes the downstream partitioned dependencies, but just for the current day? Thanks in advance.

We have about 100 time series we want to use, along with some new ones that might be created by analysts. How should we think about this, create 100 individual assets and all new ones needs to be defined in code as well sounds like the correct approach. But is there another way or methodology i am missing?

Hi Team, I have a question on partition dependencies. I have multi-dimensionally partitioned assets - consisting of day and static partition dimensions. I would like to trigger a downstream daily-partitioned asset that runs only when all upstream static partitions for that day has been materialized. Does dagster support this? The docs only mention support for partition dependencies when upstream asset and downstream asset are both time window-partitioned. What if in case where upstream is multi-partitioned (daily and static) and downstream time partitioned (daily-only)?

I'm not getting anything show up in the sensor logs?

Hi all, I hope you'r good :) One question : My goal is to be sure the connection is OK before executing extraciton assets. I tried to do it with a @CREATE_CONNECTION asset passing the connection object to @EXTRACT_ASSETS, if the connection is valid. How we can do it ? I think it's related to 'dagster type' but did'nt manage to make it work … Thanks for your help :) Example code of what i try to acheive :

@asset(key_prefix=["ETL"],compute_kind="Connection")

def CREATE_CONNECTION():

my_connection = RFC(user="xxxx",

passwd="xxxxx", system="xxxx")

return my_connection

@asset(key_prefix=["ETL"])

def EXTRACT_TABLE(CREATE_CONNECTION):

DATA = CREATE_CONNECTION.get_table(arg…)

return DATA

What is the best way to have a partitioned graph backed asset depend on the latest materialization of some other (non-partitioned) asset?

This message was deleted.

Good morning all, I have a (probably) simple question of how to disable debug level logs while running unittests. I run my unittests with

python -m unittest -v

and while it's perfectly picking up all tests and running them properly, it's a bit difficult to read due to the built in DEBUG logs that are generated on every asset creation. While I prefer to only surpress these debugs during unittests, I've tried adding the following to my `DAGSTER_HOME\dagster.yaml`:

python_logs:
  python_log_level: INFO

but that didn't help (debug messages still visible). It did disable them in dagit though (which is actually what I didn't want :)). I've also tried:

import logging
from dagster import get_dagster_logger

class TestMyTestCase(unittest.TestCase):
    ...

if __name__ == "__main__":
    logger = get_dagster_logger()
    logger.setLevel(<http://logging.INFO|logging.INFO>)
    unittest.main()

This also didn't seem to help. I'd also prefer it if I can set the logging info on one location for all unit tests, rather than every file that contains tests (as I'm seperating my tests over multiple files). Any suggestions?

Anyone know how to get around this error when initially starting Dagster?

ModuleNotFoundError: No module named 'dagster_postgres'

I am running into a weird issue and this is probably a python - pandas issue instead of a dagster issue and the only way I could get around it in pyhton was return_type being arrow and use connectorx. I have a table that has about 20 million rows in it and I am trying to hash the rows and upload them to snowflake but it is maxing out the memory (32 GB) and crashing the operation. To speed this up I went to connectorx and using arrows it now takes 2 minutes to pull and only uses 11GB of memory vs the 32GB that pandas was using but I canot figure out how to use it in dagster since it is requiring me to us Pandas. If anyone has a workaround and/or any advice I would love to hear it.

Is there a way to end a graph early if a step produces no data (and do a dynamic map otherwise)?

How do I find out information about how each run is being triggered? When I materialize all assets in dagit - some assets get materialized twice or more in separate runs. Is there a way in UI or logs to trace how dagster start each runs?

is it possible to fetch the recent Dagster Run status in the Dagster OP?

small thing: autocomplete seems to currently be broken for the new pythonic resources, showing only **data when initializing the objects rather than the properties

Running dagster as a service on an EC2 instance and am currently running it in the background. It wants to stop whenever I log out of the instance however. Currently working on figuring out how to run it as a systemd service but does anyone know how to keep it running otherwise?

Hi team, I have an asset that hasn't been materialized yet. However, other assets would like to use the input of this function. Is there a way to materialize this asset in the

load_input

function when reading it?

hey folks -- is there a way to launch the dagit UI via python script or API? I'm simply looking for the equivalent of running

dagit -f

without having to use a CLI

Hello everyone, I’m playing around with Dagster and I was wondering, is there a rule of thumbs guide of when to use a python function vs an asset with no materialization parameters? Should all the function be an asset? or should I still creates python functinos and use them in my assets?

I’ve been using dagster-serverless for a while and it’s working great. I haven’t developed locally in a while, but I’ve now I’m running into a problem. Even without any changes to the code that is running in dagster-serverless, every job I try to run locally eventually raises this error:

Multiprocess executor: child process for step was terminated by signal 7 (SIGBUS).
dagster._core.executor.child_process_executor.ChildProcessCrashException

The error comes up at different times in the pipeline, but it happens within 2 minutes every time. I’m running dagit in a container on an M1 Mac. I’ve assigned 8GB memory to docker, and it doesn’t look like I’m getting at all close to the memory limit. I’m a bit stumped here.

I'm attempting to materialize a DAG with ~500 nodes and postgres is complaining about the initial kickoff. Is this a known issue? Also, is there a set of docs/practices for large-scale deployments?

sqlalchemy.exc.OperationalError: (psycopg2.errors.ProgramLimitExceeded) index row size 3424 exceeds btree version 4 maximum 2704 for index "idx_run_tags"
DETAIL: Index row references tuple (0,27) in relation "run_tags".
HINT: Values larger than 1/3 of a buffer page cannot be indexed.
Consider a function index of an MD5 hash of the value, or use full text indexing.

[SQL: INSERT INTO run_tags (run_id, key, value) VALUES (%(run_id)s, %(key)s, %(value)s)]
[parameters: ({'run_id': 'f09f8a7d-5501-4985-8c88-3046e52e43b9', 'key': '.dagster/repository', 'value': '__repository__@user-code-location'}, {'run_id': 'f09f8a7d-5501-4985-8c88-3046e52e43b9', 'key': 'dagster/partition', 'value': '2023-03-12'}, {'run_id': 'f09f8a7d-....