Logging seems not to be emitted in sensors, e.g. in a sensor: get_dagster_logger().info("foo") emits nothing. I know I can emit a message in a SkipReason, but it's still helpful for debugging a sensor to have logging. Am I mis-configured, or is this by design?

Hello, I am trying to understand the difference between I/O managers and Dagster type loaders/materializers. I can write an I/O manager with a switch case that knows how to write to different output format based on metadata. However, dagster-pandas seems to achieve the same with a materializer. When does it make sense to use one or the other?

hi @Bit Unwise! I think "factory function to build assets" is your best bet -- asset config can change the functionality, but not the identity of an asset (so if you reconfigure an asset it'll still have the same unique asset key). It sounds like you just need

F(endpoint, schema, some config) -> List[AssetsDefinition]

which you can apply multiple times with different parameters. I think this pattern would end up pretty clean, but happy to talk through details if there are sticking points

I've created a feature request here but I've since dug into the docs and found that I might be able to use MetadataValue for what I need. Is there any way to use the

MetadataValue

class to emit compiled SQL and have it print nicely in the asset catalog?

👋 Is the Dagster

DagsterUserCodeUnreachableError

indicative of a communication error between the User Code and the application, or could it be syntax?

Hey Dagster team! Looking for some advice on how to structure a job, can you help? This job pulls data from multiple database shards and simply merges that data together. I'm currently accomplishing this by creating a different op for each shard. I would like to avoid repeating the same op code for each shard but also maintain the ability to fetch data concurrently from the different shards. Here's an example of my code:

@op(
    description="Get data shard1",
    required_resource_keys={"db_resource1"}
)
def get_data_shard1(context):
    query = "select * from table"
    db = context.resources.db_resource1
    df = db.fetch_data_by_query(query)
    return df

@op(
    description="Get data shard2",
    required_resource_keys={"db_resource2"}
)
def get_data_shard2(context):
    query = "select * from table"
    db = context.resources.db_resource2
    df = db.fetch_data_by_query(query)
    return df

@op(description="Merge cross shard data")
def merge_data(context, df1, df2):
    merged = pd.concat([df1, df2])
    return merged

@job(
    resource_defs={
        "db_resource1": db_resource.configured({"database": "shard1"}),
        "db_resource2": db_resource.configured({"database": "shard2"}),
    }
)
def multi_shard_data_job():
    df1 = get_data_shard1()
    df2 = get_data_shard2()
    merged = merge_data(df1, df2)

Thanks for reading!

Hi! I have an op that has an unconnected input along with some connected ones. In the job, do I just pass the others ones by name and leave that one blank?

@job
def my_job()
    op_many_inputs(connected1=input1, connected3=input2)

@op
def op_many_inputs(connected1:str, unconnected2: int, connected3: int):

Hi, new to dagster and trying to learn more about software defined data assets. One thing I am not sure I quite understand is how passing data as input to an asset works. It appears as though the data is passed to the asset in-memory after being de-pickled, but what if that data is too large to fit in memory? Suppose I have a dataset of 1M images (an array of s3 URLs) that I want to apply some transformation to. Would I pass this array of s3 URLs into my asset, and load each image from s3 in the asset itself? It feels like this goes against the notion of separating io from business logic. Would the right move to be to partition the 1M images? Furthermore, suppose you want to implement an asset which trains a model, and the images live in s3, how would you implement that? Would be cool to see an example of a model training with a lot of data, and how that's implemented in this new data asset paradigm One other question, can you have a graph of assets, where one asset might run inside of a docker container from image A, and another runs inside of a docker container from image B? Is that currently possible?

Hello team! I'm trying to implement

@slack_on_failure

on one of my dags, and I want to customize the message, Is there a way to obtain the full stack trace from a HookContext? If I try with `context.op_exception`; it just prints the error but not the full stack trace that I can see in the UI. For example:

'bool' object is not subscriptable

But I want something like this:

dagster.core.errors.DagsterExecutionStepExecutionError: Error occurred while executing op "exception_op":
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_plan.py", line 222, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 343, in core_dagster_event_sequence_for_step
    _step_output_error_checked_user_event_sequence(step_context, user_event_sequence)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 69, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/compute.py", line 169, in execute_core_compute
    for step_output in _yield_compute_results(step_context, inputs, compute_fn):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/compute.py", line 146, in _yield_compute_results
    user_event_generator,
  File "/usr/local/lib/python3.7/site-packages/dagster/utils/__init__.py", line 400, in iterate_with_context
    return
  File "/usr/local/lib/python3.7/contextlib.py", line 130, in __exit__
    self.gen.throw(type, value, traceback)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/utils.py", line 78, in solid_execution_error_boundary
    ) from e
The above exception was caused by the following exception:
TypeError: 'bool' object is not subscriptable
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/utils.py", line 47, in solid_execution_error_boundary
    yield
  File "/usr/local/lib/python3.7/site-packages/dagster/utils/__init__.py", line 398, in iterate_with_context
    next_output = next(iterator)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/compute_generator.py", line 65, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "/opt/dagster/app/batch/dagster_error_test.py", line 25, in exception_op
    test = context.op_config["exception"]

Thanks!

is it possible to see how a certain resource is configured in the dagit UI? trying to use .env file along with python-dotenv but having trouble configuring my resources in a production vs local environment. It appears that the wrong set of resources are being used.

Hi all, could anyone be able to help me with this error I am encountering: "Missing required config entry "ops" at the root.". I have tried to pass a config file through but I am having no luck. Can anyone help?

Hi all, what would be the smartest / simplest way for an asset to pass settings to its IOManager? I’m currently learning Dagster by creating a simple IOManager for ClickHouse and want to pass database and table definitions from an asset. The closest to what I think I need from the docs was via context.name but what if I want to pass more data? Would setting metadata in

@asset(metadata={'database':'db','table':'tab'})

And then fetching it through

context.metadata['database']

be a good idea?

Hi, is there a way to return a result (json) from a job, so that it could be later grabbed by a GraphQL query after the job completes? Or resources is the only way to return some information to the outside?

hi team, we want to upgrade to 1.0 but some packages we’re using are still behind (and it looks like 1.0.5 was yanked) • https://pypi.org/project/dagster-dbt/#history • https://pypi.org/project/dagster-aws/#history any idea on when these packages will be compatible with 1.0?

Hi all — Is there a way to run a job with sequential ops, where all ops don’t have outputs? I’m not sure how to do that since I’ve only been passing outputs to create a sequential op pipeline Ex: op1 => print(‘hello’) op2 => print(‘how are you?’) op3 => print(‘have a nice day’) where job would run op1, then op2, then op3. **Note that all ops don’t have a output variable that I can pass to the next one

If I connect to the graphql endpoint via websocket, should I expect to see all pipeline events here or are there specific channels I need to subscribe to?

Hi @owen I'm trying to construct an op for dbt deps like this

@op(required_resource_keys={"dbt"})
def dbt_deps_op(context):
    context.resources.dbt.cli("deps")

I am using rpc-server and I'm getting this error on the dbt server "API Exception: {'type': 'InternalException', 'args': ('No matching handler found for rpc method deps (which=deps)',) This is my resource

test_dbt_rpc_sync_resource = dbt_rpc_sync_resource.configured({
    "host": { "env": "DBT_HOST" },
    "port": { "env": "DBT_PORT" }
})

Is there a programmatic way to clean the

dagster_home/storage

folder? Ideally I would like to remove all data associated with runs >= 10 days old.

Hello, I would like to inquire about the SOC 2 Type 1 report. I have checked the following website and sent an email to sales@elementl.com on September 13, but have not received a reply. https://dagster.io/blog/soc2-compliance-dagster-blog Can you please give me a copy of your SOC 2 certification?

Hey all! I am building a scheduled job, that should detect long running jobs, and notify the user in such a case. In the dagster-support channel I was able to find valuable resources (see https://dagster.slack.com/archives/C01U954MEER/p1628717201346100?thread_ts=1628715859.344300&cid=C01U954MEER ) I decided against using a

RunStatusSensor

as I seem to be able to inspect only one job-state at a time, but I would need to monitor multiple jobs’ runtimes. This issue I run into with the code example below, is that I am not able to retrieve the run_stats for a job. No error is being logged and no code seems to be executed after trying to retrieve run_stats via the DagsterInstance found in the op-context. What could that issue be here? btw, I am running dagster version

1.0.7

from dagster import DagsterInstance, DagsterRun

@op()
def op_notify_long_runtime_jobs(context):
    instance: DagsterInstance = context.instance
    runs: Iterable[DagsterRun] = filter(
        lambda run: run.status == DagsterRunStatus.STARTED and
                    run.job_name != 'notify_long_runtime_jobs',
        instance.get_runs())

    def evaluate_run_time(r):
        print(f"{r.job_name} | {r.run_id} | {r.status} | {type(r)}") # anything below, is not being printed
        print(f"{instance.get_run_stats(r.run_id)}") 
        run_stats = instance.get_run_stats(r.run_id) 
        elapsed_time = time.time() - run_stats.start_time

        if elapsed_time > 20:  # TODO Configurable Limit in seconds
            print(f"Long running job {r.run_id}! Run for {elapsed_time} seconds ")  # TODO run slack notification

    [evaluate_run_time(r) for r in runs]


@job()
def notify_long_runtime_jobs():
    op_notify_long_runtime_jobs()

Hello, is there a way to automatically add a key prefix to all assets in a given repository and repository location, without having to add it to each individual asset? Say, for all assets in

repoA

at repository location

location

I'd like to pre-pend

location

,

repoA

to

prefix_list

. I use different repository locations for different branches of my code (e.g., one repository location for the user code deployment for

develop

, one repository location for the user code deployment for

feature-1234

, etc.), and it'd be great to allow these to coexist easily. I've seen utilities like prefix_assets but I am wondering whether there's a way to do this automatically for all repositories, taking the location into account as well.

Hello, I’ve been looking through the documentation and slack history, but I couldn’t find any advice on if this was possible. Is it possible to have an “overloaded” asset? That is, one that has multiple associated ops (or is this a major anti-pattern?). I have an asset that needs to be filled as a part of a normal pipeline on a weekly cron, but more frequently deletes users from that table (GDPR type stuff). Schema doesn’t change, obviously - but I can’t run the pipeline more frequently due to other business requirements. Alternate would be to have a microservice that amends the table, but I would like to keep everything in dagster if possible.

hi, if I want some Job to start a docker container to execute some application script, that returns a text file. should I consider the container as an asset or a resource?

Hi, I am trying to supply

run_config

to a schedule function. My set up looks like this:

pipelines
├── data_pipeline
│   ├── __init__.py
│   ├── repo.py
|   ├── schedules.py
|   ├── config
│       ├── run_config.yaml

The following

schedule

is defined in the

schedules.py

file:

@schedule(job=update_db_metadata_job, execution_timezone='Africa/Johannesburg', cron_schedule="08 14 27 * *")
def metadata_schedule_dev():
    return RunRequest(
        run_key=None,
        run_config=config_from_files(
            [file_relative_path(__file__, "/config/run_config.yaml")]
        ),
    )

However, when the schedule executes, I get the following error about not being able to produce a result:

dagster._core.errors.DagsterInvariantViolationError: File or glob pattern "/config/run_config.yaml" for "config_files" produced no results.
  File "/usr/local/lib/python3.8/site-packages/dagster/_grpc/impl.py", line 267, in get_external_schedule_execution
    return schedule_def.evaluate_tick(schedule_context)
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/definitions/schedule_definition.py", line 500, in evaluate_tick
    result = list(ensure_gen(execution_fn(context)))
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/definitions/decorators/schedule_decorator.py", line 139, in _wrapped_fn
    result = fn()  # type: ignore
  File "/dagster/dhis2_pipeline/schedules.py", line 32, in metadata_schedule_dev
    run_config=config_from_files(
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/definitions/utils.py", line 153, in config_from_files
    raise DagsterInvariantViolationError(

I have tried several variations of supplying the config via a file but can't get it to work. A similar method used to work when I supplied the config as above with

config_from_files

and using

file_relative_path

to the job but I want to move the config to be defined in the schedule rather. Any suggestions how to do this?

Hi, I am trying to call an op with a dynamic output within an asset, however, when using the .collect( functionality, this error appears: 'generator' object has no attribute 'collect'. Has anyone seen this before and if so is there a solution? In future I will also use the .map() function so if anyone has run into a similar error using that it would be helpful as well. Thank :)

How to do parallel processing (for op functions) using python run ?

Hi all. Is it possible to have an input parameter of type

Dict

on a job level? I wonder if I'm doing something wrong, since when when I'm trying to run this job:

from typing import Any, Dict

from dagster import job, op
from dagster_aws.s3 import s3_pickle_io_manager, s3_resource


@op
def test_op(context, parameters_dict: Dict[str, Any]):
    <http://context.log.info|context.log.info>(parameters_dict)


@job(resource_defs={
    "io_manager": s3_pickle_io_manager,
    "s3": s3_resource,
})
def test_job(parameters_dict: Dict[str, Any]):
    test_op(parameters_dict)

With the following parameters:

inputs:
  parameters_dict:
    param_1: value1
    param_2: 2
resources:
  io_manager:
    config:
      s3_bucket: my-bucket

I'm getting the following error:

dagster.core.errors.DagsterTypeLoadingError: Error occurred while loading input "parameters_dict" of step "test_op":
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_plan.py", line 224, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/execute_step.py", line 316, in core_dagster_event_sequence_for_step
    step_input.source.load_input_object(step_context, input_def)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/inputs.py", line 627, in load_input_object
    return dagster_type.loader.construct_from_config_value(step_context, config_data)
  File "/usr/lib/python3.8/contextlib.py", line 131, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/errors.py", line 191, in user_code_error_boundary
    raise error_cls(
The above exception was caused by the following exception:
AttributeError: 'str' object has no attribute 'items'
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/errors.py", line 184, in user_code_error_boundary
    yield
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/execution/plan/inputs.py", line 627, in load_input_object
    return dagster_type.loader.construct_from_config_value(step_context, config_data)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/python_dict.py", line 38, in construct_from_config_value
    runtime_value[key] = self._value_dagster_type.loader.construct_from_config_value(
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/config_schema.py", line 156, in construct_from_config_value
    return self._func(context, config_value)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/builtin_config_schemas.py", line 52, in _any_input_schema
    return load_type_input_schema_dict(config_value)
  File "/home/gsterin/.local/lib/python3.8/site-packages/dagster/core/types/builtin_config_schemas.py", line 35, in load_type_input_schema_dict
    file_type, file_options = list(value.items())[0]

And with scalar inputs the same code works just fine

trying to put together some a poc using dagster and databricks I have about 40ish assets that I'm attempting to materialize and from what it looks like a random set of them will fail with the following error. Looks to be some timing related thing with how assets are materialized and recording? The same asset doesn't always fail when I materialize the set of them.

FileNotFoundError: [Errno 2] No such file or directory: '/home/leaf/code/origo-analytics/astra/tmpa7a0jq5q/history/runs/dc112b44-3e6a-4c89-a114-657d28380d17.db-wal'
  File "/home/leaf/.cache/pypoetry/virtualenvs/astra-DF50hgcZ-py3.8/lib/python3.8/site-packages/dagster/_core/execution/plan/execute_plan.py", line 224, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home/leaf/.cache/pypoetry/virtualenvs/astra-DF50hgcZ-py3.8/lib/python3.8/site-packages/dagster_databricks/databricks_pyspark_step_launcher.py", line 193, in launch_step
    self._upload_artifacts(log, step_run_ref, run_id, step_key)
  File "/home/leaf/.cache/pypoetry/virtualenvs/astra-DF50hgcZ-py3.8/lib/python3.8/site-packages/dagster_databricks/databricks_pyspark_step_launcher.py", line 398, in _upload_artifacts
    build_pyspark_zip(zip_local_path, self.local_dagster_job_package_path)
  File "/home/leaf/.cache/pypoetry/virtualenvs/astra-DF50hgcZ-py3.8/lib/python3.8/site-packages/dagster_pyspark/utils.py", line 21, in build_pyspark_zip
    zf.write(abs_fname, os.path.relpath(os.path.join(root, fname), path))
  File "/usr/local/lib/python3.8/zipfile.py", line 1741, in write
    zinfo = ZipInfo.from_file(filename, arcname,
  File "/usr/local/lib/python3.8/zipfile.py", line 523, in from_file
    st = os.stat(filename)

Are there recommended upper bounds for a

QueuedRunCoordinator

max_concurrent_runs

? We are using

EcsRunLauncher

backed by a fargate cluster so our compute is flexible

Hi Team, want to know how read dbt.freshness result/log. Purpose of this is to know the result of Dbt freshness check invoked from Dagster. Urgent.