Hi 👋 I am looking into the Spark integration documentation so that I can launch spark jobs on AWS EMR. Do I understand correctly that there is no OP which will launch an EMR cluster and then run a step on it, and that instead the EMR cluster should already be running and Dagster can add / launch a step on that cluster? I am looking for something similar to Airflow's EmrCreatejobFlowOperator.

Hello dagsters! The graphql api allow to retrive some asset value? It would be a entrypoint to https://docs.dagster.io/_apidocs/repositories#dagster.RepositoryDefinition.load_asset_value Actually, the dagster has a lot of metadata/functionalities around each asset... Wouldn't it be great to have a graphql schema generated out-of-the box to query the assets? Is it dagster or some third-party working in something along these lines?

Hi, how should I go about solving this error?

Sensor daemon caught an error for sensor git_comment_sensor : dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code server

Stack Trace:
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_daemon/sensor.py", line 476, in _process_tick_generator
    yield from _evaluate_sensor(
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_daemon/sensor.py", line 540, in _evaluate_sensor
    sensor_runtime_data = repo_location.get_external_sensor_execution_data(
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_core/host_representation/repository_location.py", line 808, in get_external_sensor_execution_data
    return sync_get_external_sensor_execution_data_grpc(
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_api/snapshot_sensor.py", line 54, in sync_get_external_sensor_execution_data_grpc
    api_client.external_sensor_execution(
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_grpc/client.py", line327, in external_sensor_execution
    chunks = list(
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_grpc/client.py", line124, in _streaming_query
    raise DagsterUserCodeUnreachableError("Could not reach user code server") from e

The above exception was caused by the following exception:
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
        status = StatusCode.DEADLINE_EXCEEDED
        details = "Deadline Exceeded"
        debug_error_string = "{"created":"@1667927058.507662760","description":"Deadline Exceeded","file":"src/core/ext/filters/deadline/deadline_filter.cc","file_line":81,"grpc_status":4}"
>

Stack Trace:
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/dagster/_grpc/client.py", line122, in _streaming_query
    yield from response_stream
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/grpc/_channel.py", line 426, in __next__
    return self._next()
  File "/home/u00ubc1kg5n2YAppSj357/miniconda3/envs/rapids-22.10/lib/python3.9/site-packages/grpc/_channel.py", line 826, in _next
    raise self

Limiting Concurrency of Dynamic Ops Problem: One of my ops spawns n ops of another type. I'd like to limit the concurrency of just these types of ops as they all compete for the same resource. I've looked at this documentation, but this seems to limit concurrency in separate runs, not within the run. Question: Is there a way, via tags or otherwise, to limit the number of DynamicOut?

Simple Q: How do I define the relationship between two ops, especially if I am only defining one op (other is imported from dagster_dbt). For example:

@op
def upload_docs_to_s3():
    foo

@job
def dbt_cli_docs():
    dbt_docs_generate_op()
    upload_docs_to_s3()

In this case, both ops try to run at the same time. In any case, is there a simple way to set relationship of ops in general? Like:

op1 >> op2 >> op3

Small question, when you use the config_schema Field() function, to specify default, and description. Where does the description actually show up in the Dagit? I was hoping to scaffold missing config and have there be a description somewhere saying "Fill in this field or it will run with default value".

This issue (https://github.com/dagster-io/dagster/issues/9439) implies that container logs from

k8s_job_op

runs should be available in dagit after a job has completed. I’ve never seen this happen in practice with any of my

k8s_job_op

usage and I always get the blank loading screen. Is there any specific configuration that’s required to make this work?

Hi I am trying to get a scheduled job running. It follows the very first “schedule” example in terms of simplicity. As it starts executing I receive the following error in the console. Nothing indicating a run appears in my daemon terminal or the dagit terminal. I can’t begin to figure out where to set the path

root:config:base_dir

value.

dagster._core.errors.DagsterInvalidConfigError: Error in config for resource io_manager
Error 1: Value at path root:config:base_dir must not be None. Expected "(String | { env: String })"
  File "/Users/crawford.collins/projects/hello-flow/.venv/lib/python3.9/site-packages/dagster/_grpc/impl.py", line 412, in get_external_execution_plan_snapshot
    create_execution_plan(
  File "/Users/crawford.collins/projects/hello-flow/.venv/lib/python3.9/site-packages/dagster/_core/execution/api.py", line 1058, in create_execution_plan
    resolved_run_config = ResolvedRunConfig.build(pipeline_def, run_config, mode=mode)
  File "/Users/crawford.collins/projects/hello-flow/.venv/lib/python3.9/site-packages/dagster/_core/system_config/objects.py", line 209, in build
    config_mapped_resource_configs = config_map_resources(resource_defs, resource_configs)
  File "/Users/crawford.collins/projects/hello-flow/.venv/lib/python3.9/site-packages/dagster/_core/system_config/objects.py", line 286, in config_map_resources
    raise DagsterInvalidConfigError(

because the error message cuts off, I can’t figure out where my error is coming from or even begin to fix it

What is the current best practice for reexecuting a pipeline? I want to restart my pipeline from a specific step that failed. I can find a description on Stack Overflow here , but I cannot import

reexecute_pipeline

from dagster 1.0.15, which makes me think there is now another way?

Hello team! I have a question about

dagster + dbt

. Following the documentation I'm a bit confused about how to configure the

dbt

command. I managed to wire up a

dbt_cli_resource

which reads from the configuration file (which is different by environment). But then I need to call

load_assets_from_dbt_project

to get the asset for my job, which happens at import time (and not runtime). How do I access the

project-dir

and

profiles-dir

settings from the

dbt

resource at this stage? I'm trying to avoid hardcoding those paths in two places and would prefer to only have them in the config file under the

dbt

resource

Thread: Partitioned multi_asset

the CLI currently doesn’t support specifying op selection. do you mind filing a feature request issue to help prioritize?

Is there a way to skip a scheduled job run if a prior run is still running?

Hi! I'm upgrading a service from 0.12 to 1.0.* and can't find what replaced

fs_file_cache

from

dagster.core.storage.file_cache

, all I can see is that it was removed in 0.15

Hi, I could see this in the changes log for 1.0.17, but couldn’t find any document for it. My interpretation of the text is that, similar to multi-asset, we can now have a run which reads multiple partitions of upstream assets, and generate multiple partitions of the downstream asset. Is there any more details documents and/or an example for this cool new feature? Thanks.

You can now launch a run that targets a range of asset partitions, by supplying the "dagster/asset_partition_range_start" and "dagster/asset_partition_range_end" tags.

Hey all - Wondering if anyone here has made an attempt at "asset versioning". We are wondering how to: 1. see which partitions for a given asset differ in version (perhaps the underlying dbt model sql changed) 2. determine if a downstream asset's partition is "out of date" compared to an upstream asset's same partition

Hi there, just a simple question: What is the difference between multi_asset and a factory of assets. Why would I use the first instead of the second?

Hi, guys, sorry to bother you with such a simple question. But does anyone know how to create a test that executes the load repository to verify if it has any errors loading the source code? I got 3 repositories, and when I'm deploying something, I would like to test on the pipeline if the repo was correctly loaded before doing the deployment, so I can avoid seeing the error just when it deploys.

How might I combine

dagster.dynamic_partitioned_config

with

dagster.ConfigMapping

? https://docs.dagster.io/_apidocs/partitions#dagster.dynamic_partitioned_config https://docs.dagster.io/_apidocs/config#dagster.ConfigMapping

Is it possible to add a partition when a job is created? If I had a job that had a dynamic partitioning scheme that was based off of buckets sub-directories that exist, and the job can/will create one of those sub-directories on its first run, can the job when it runs add an additional key to that dynamic partition?

How can I manage users for Dagit UI access in an Dasgter open source instance?

For the toy example of:

python
@asset
def upstream_nonpartitioned_1():
    return 1

@asset
def upstream_nonpartitioned_2():
    return 1

@asset
def merged_nonpartitioned(upstream_nonpartitioned_1, upstream_nonpartitioned_2):
    return upstream_nonpartitioned_1 + upstream_nonpartitioned_2

merged_nonpartitioned_job=define_asset_job("non_partitioned", selection="ea/merged_nonpartitioned")

@multi_asset_sensor(
    asset_keys=[AssetKey("ea/upstream_nonpartitioned_1"), AssetKey("ea/upstream_nonpartitioned_2")],
    job=merged_nonpartitioned_job,
    default_status=DefaultSensorStatus.RUNNING,
)
def asset_a_and_b_sensor_with_skip_reason(context):
    asset_events = context.latest_materialization_records_by_key()
    print('***')
    print(asset_events)
    print(asset_events.values())
    print('***')
    if all(asset_events.values()):
        context.advance_all_cursors()
        return RunRequest()
    elif any(asset_events.values()):
        materialized_asset_key_strs = [
            key.to_user_string() for key, value in asset_events.items() if value
        ]
        not_materialized_asset_key_strs = [
            key.to_user_string() for key, value in asset_events.items() if not value
        ]
        return SkipReason(
            f"Observed materializations for {materialized_asset_key_strs}, "
            f"but not for {not_materialized_asset_key_strs}"
        )
    else:
        return SkipReason(f"NO materializations for {asset_events}, ")

assets_list = load_assets_from_package_module(event_assets, group_name="event_assets", key_prefix="ea")

@repository
def assets_modern_data_stack():
    return[assets_list, 
        asset_a_and_b_sensor_with_skip_reason,
        ]

Always no asset materializations are logged form the dagster daemon logs:

{AssetKey(['ea/upstream_nonpartitioned_1']): None, AssetKey(['ea/upstream_nonpartitioned_2']): None}
dict_values([None, None])

Even after manually materializing each of the upstream assets, the materialization events are not picked up.

hi, I am starting to take a look at dagster and was wondering how to debug code locally in my IDE (PyCharm). is there an article which explains the setup? thx 🙏

Hello all!

Can anyone tell me how to match the dagster and dagit versions in the local environment?

Hi all, i'm currently getting this error on graph asset creation Is it caused by the optional branches?

dagster._check.CheckError: Invariant failed. Description: All leaf nodes within graph 'branch_fetch_waterfall' must generate outputs which are mapped to outputs of the graph, and produce assets. The following leaf node(s) are non-asset producing ops: {'google_bigquery_upload_json_7', 'google_bigquery_upload_json_9', 'google_bigquery_upload_json_3', 'google_bigquery_upload_json_8', 'google_bigquery_upload_json_4', 'google_bigquery_upload_json_10', 'google_bigquery_upload_json_2', 'google_bigquery_upload_json_5', 'google_bigquery_upload_json_6', 'google_bigquery_upload_json'}. This behavior is not currently supported because these ops are not required for the creation of the associated asset(s).

Hey all, I got this alert,

unloadable schedules

, I didn't see any problems with the code (repository or schedule) also another jobs in same repository worked fine. Any idea?

Hello! dagsir how do a custom ressource i've made get access to an InputContext? should i simply pass it into each function of the resource as done in a IOManager?

Hello, dagster spin team! I have a question about using of sensors, for instance:

@run_status_sensor

. There is parameter

request_jobs: Optional[Sequence[Union[GraphDefinition, JobDefinition]]] = None

. Is it possible to pass two (or more) same jobs with different config? How should I config it? Config for one job is pretty simple:

"resources": {
    "first_resource": {
        "config": {
            "nice_param": "a"
        }
    }
},

But how can I define second one?

also btw how do you show sample data in dagster? is that through meta data? i.e a pyspark dataframe