on a related note to the above, is there an easy way to run a backfill on a schedule? • I have a data source that is sensibly partitioned into days but is updated in a “lumpy” way, i.e. not daily but X days at a time, every X days, with a variable number X. • I would like to use asset sensors to then kick off downstream jobs upon materialization The obvious way to handle this (for me) would be to just attempt a backfill every day, but I don’t see an easy way to do that within the scheduler code. I’d like to avoid programmatically making calls to the dagster CLI, e.g. defining a cron job that just runs

dagster job backfill

from console every day, but if that’s the only way to do it then i guess I can do that

I’m looking for some clarification on how configs work. I have an asset that defines two config parameters via the config_schema. I then create a schedule, which has a job using ‘define_asset_job’ which has a config that is ultimately loaded from a yml file on disk. I then have a repository.py file that I load into dagster. When my schedule executes, I am expecting the schedule to execute the job, which will load the yml from disk and materialize my asset. However, the schedule errors stating that there is no config. Am I correct in expecting the yml to be loaded from disk? Or does dagster intervene and expect the config to be supplied via the UI?

Hey there, I am using the create_shell_command_op to run some node cron jobs. My current issue is that I want to log the output of op while the job runs. It only logs the output when the job finishes as one big blob of text. Is there a way around this?

Is there a way to mix partitioned and non-partitioned assets in a job? E.g., I want to get the most recent partition for a source asset before running the non-partitioned transformation that generates a downstream asset.

When I try to call

context.get_step_execution_context()

on a directly invoked op with

build_op_context

, I get this error:

dagster.core.errors.DagsterInvalidPropertyError: The get_step_execution_context methods is not set on the context when a solid is directly invoked.

Is there a "directly-invoked" flag on the

context

object?

I'm hoping there's some way to tell when my op has been directly invoked, so I can avoid using certain

context

properties

I have an op that generates a

DynamicOutput

with

for val in ['cat', 'dog', 'panda']:
        yield DynamicOutput(
            complex_fn(val),
            mapping_key=val,
        )

Downstream, I'd like to same an output to s3 with the name

f"output_{val}"

using the mapping_key. Is this possible? If not, what pattern should I be using to access

val

later on? What I'm trying to do in this workflow is (a) make pandas plots based on

n

dataframes (this is the dynamic part) and then, (b) save each of those plots to a unique file on s3. Alternatively, is there a way I could merge the plots back into a list to use with the original list used to break them apart?

Hi All, our team has been getting the following error message frequently from our sensors / schedules as of late, we are curious as too what could potentially be causing it.

2022-11-01 12:02:44 +0000 - dagster.daemon.SchedulerDaemon - INFO - Evaluating schedule `hive_metastore_sync_gl_schedule` at 2022-11-01 07:00:00 -0500
    raise self
  File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 826, in _next
    return self._next()
  File "/usr/local/lib/python3.7/site-packages/grpc/_channel.py", line 426, in __next__
    yield from response_stream
  File "/usr/local/lib/python3.7/site-packages/dagster/_grpc/client.py", line 122, in _streaming_query
Stack Trace:
>
	debug_error_string = "{"created":"@1667304153.030278088","description":"Deadline Exceeded","file":"src/core/ext/filters/deadline/deadline_filter.cc","file_line":81,"grpc_status":4}"
	details = "Deadline Exceeded"
	status = StatusCode.DEADLINE_EXCEEDED
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
The above exception was caused by the following exception:
    raise DagsterUserCodeUnreachableError("Could not reach user code server") from e
  File "/usr/local/lib/python3.7/site-packages/dagster/_grpc/client.py", line 124, in _streaming_query
    sensor_execution_args
  File "/usr/local/lib/python3.7/site-packages/dagster/_grpc/client.py", line 305, in external_sensor_execution
    cursor=cursor,
  File "/usr/local/lib/python3.7/site-packages/dagster/_api/snapshot_sensor.py", line 61, in sync_get_external_sensor_execution_data_grpc
    cursor,
  File "/usr/local/lib/python3.7/site-packages/dagster/_core/host_representation/repository_location.py", line 796, in get_external_sensor_execution_data
    state.instigator_data.cursor if state.instigator_data else None,
  File "/usr/local/lib/python3.7/site-packages/dagster/_daemon/sensor.py", line 508, in _evaluate_sensor
    sensor_debug_crash_flags,
  File "/usr/local/lib/python3.7/site-packages/dagster/_daemon/sensor.py", line 457, in _process_tick_generator
Stack Trace:
2022-11-01 12:02:41 +0000 - dagster.daemon.SensorDaemon - ERROR - Sensor daemon caught an error for sensor slack_on_run_failure : dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code server

We are using the Helm Chart to deploy so curious as too how it could be losing connection.... I do see that DEADLINE_EXCEEDED message which I thought was due to a sensor going over the sensor alive time limit but not sure if that is still related..... Any help appreciated.

hey yall, does dagster-dbt currently load tests into the DAG? Looking at the source I see:

deps = _get_deps(dbt_nodes, selected_unique_ids, asset_resource_types=["model"])

Which would seem to only load models

hi there, I’m interested in merging a dagster postgres database into another pre-existing dagster postgres database. Are there any docs/tips/checklists that can help with this? My current plan is: 1. try ensure they are on the same alembic version 2. pg_dump old db 3. insert the contents into the target database

Trying to figure out the best way to add logging for code that doesn't have easy access to a context variable. I noticed some weird behavior when I tested get_dagster_logger() in a graph though. When I was running dagit locally, debug/info level messages failed to show up at all, while critical messages printed out to terminal as soon as dagit started (so before I had selected to run any job from the repo.py in the dagit ui). Running the graph via a job, did not reprint the messages into the dagit logging ui panel nor did they get replicated for each run through. When doing the same thing within op code or python code that the ops call, the logging behavior was exactly as I would expect: in the dagit ui, all levels of logging shown (provided toggles in the ui are set) and printed out for each run of the op. Can someone help me figure out why the behavior differs between the two?

Hey y'all I'm looking for a way to get the time when a run started, I'm guessing somehow through the context variable, to store in a database and use to correlate related data in several materializations. Is there a way to get this pretty easily (and convert it to UTC / ISO)?

In the dagit UI I see "Timing" next to runs, is there a context.run_config['timing'] or something like that?

Hello Dagsters, I would like to have a

job

that has an

asset

as input (currently the

job

is defined using

@graph

and

.to_job

) Is it possible? How?

Hello, I've created a class in which I'm trying to return some objects from it inside a op, but I'm getting this error while trying to pass their attribute values inside another step: AttributeError: 'InvokedSolidOutputHandle' object has no attribute 's3_path' where s3_path it's an attribute of my class Am I missing something? I have a class Foo() which has a lot of attr and I also have this op @op def i_will_return_class_objects(): return Output(class_object, "co1"), Output(class_object_2, "co2") and my job it's like this: @job def my_job(): var_for_object_1, var_for_object_2 = i_will_return_class_objects() insert_this_guy_at_somewhere(var_for_object_1.s3_path)

I am trying to write a custom logger. The logs are in the following format: {"status":"info","time":1667335824326,"time_µs":358 ... } However they are all being logged as info from dagster perspective even when the status is error. The examples for creating a custom logger have you create the class with level before you know the text. Is there a way to create a custom logger that could dynamically change the level of the logs based on the text in the log?

Is there a way to call @graphs from within @assets?

from dagster import graph, asset
@graph
def trial_graph():
    pass

@asset
def trial_graph_in_asset():
    trial_graph()

dagster._core.errors.DagsterInvariantViolationError: Attempted to call graph 'trial_graph' outside of a composition function. Invoking graphs is only valid in a function decorated with @job or @graph.
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\execution\plan\execute_plan.py", line 225, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\execution\plan\execute_step.py", line 360, in core_dagster_event_sequence_for_step
    for user_event in check.generator(
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\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 "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\execution\plan\compute.py", line 174, in execute_core_compute
    for step_output in _yield_compute_results(step_context, inputs, compute_fn):
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\execution\plan\compute.py", line 142, in _yield_compute_results
    for event in iterate_with_context(
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_utils\__init__.py", line 430, in iterate_with_context
    next_output = next(iterator)
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\execution\plan\compute_generator.py", line 74, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "C:\tau_dev\apollo\apollo\flux\assets.py", line 86, in trial_graph_in_asset
    trial_graph()
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\definitions\node_definition.py", line 199, in __call__
    return PendingNodeInvocation(
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\definitions\composition.py", line 377, in __call__
    assert_in_composition(node_name, self.node_def)
  File "C:\ProgramData\Anaconda3\envs\dagster_env\lib\site-packages\dagster\_core\definitions\composition.py", line 125, in assert_in_composition
    raise DagsterInvariantViolationError(

Hey all - I’m trying to create a graph-backed software-defined asset such that I can replay portions of the asset job if there are failures. The graph and SDA are set up as such:

@op
def compute(context: OpExecutionContext,) -> str:
    return "abcdef"

@op
def more_compute(
    context: OpExecutionContext,
    compute_id: str
) -> tuple[str, str]:
    if context.run.previous_run_id:
        return compute_id, "hijklm"

    raise ValueError("try again")

@graph
def compute_graph():
    return more_compute(compute())

compute_asset = AssetsDefinition.from_graph(compute_graph)

assets = with_resources(
    load_assets_from_modules(...), # contains compute_asset
    resource_defs={
        "env": env,
        "s3": s3_resource,
        "io_manager": s3_io_manager, #PickledObjectS3IOManager
    }

However, trying to replay the downstream

more_compute

op after a failure results in a warning and an error:

WARNING
No previously stored outputs found for source StepOutputHandle(step_key='compute_graph.compute', output_name='result', mapping_key=None). This is either because you are using an IO Manager that does not depend on run ID, or because all the previous runs have skipped the output in conditional execution.

STEP_FAILURE
dagster._core.errors.DagsterInvariantViolationError: Attempting to access run_id, but it was not provided when constructing the OutputContext

Is this op replay supported in graph-backed assets?

I'm confused with RunRequest https://docs.dagster.io/_apidocs/schedules-sensors#run-requests • Shouldn't the parameter asset_selection accept a value of type AssetSelection? How can I specify e.g. to select a certain AssetGroup? • Also there is the concept of "Op Selection" https://docs.dagster.io/concepts/ops-jobs-graphs/job-execution#specifying-op-selection ◦ How does it relate to the asset_selection parameter? ◦ Is it possible to pass an Op Selection via the RunRequest? Context: I'm trying to materialize a subset of Assets of an "all assets job" due to this performance issue https://github.com/dagster-io/dagster/issues/10236

Hi , Is there any way I can create a lineage between jobs and external data-sources?

Hi team! I'm running some job which looks like this:

initial_data = initial_op()

data1 = processor1_op(initial_data)
persist1_op(data1)

data2 = processor2_op(initial_data)
persist2_op(data2)

I was expecting that the order of ops will remain the same as in job definition, but Dagster run them like this:

initial_op -> processor1_op -> processor2_op -> persist1_op -> persist2_op

, which is kind of bad: results of

processorX_op

will be stored in io_manager (in memory) till the very end of the job execution, but that memory could be released probably if the op execution order was preserved. Is there any way to set some execution ordering strategies or priorities? I know there are "nothing" dependencies, but this is a bit different situation as there is no implicit dependency.

Hello, I have a question about dagster 0.13.13, on the repository I have for the same job, 2 schedules , one old hourly partitioned schedule and a second with new cron style, the problem is if I disable the hourly partitioned schedule from web interface, this schedule still run. I miss something ?

Hello! In some cases, I want to use an op in a graph, but with one input hardcoded. How do I do this? It doesn't seem like I can pass a string in the graph definition. Thanks!

@graph
def my_graph(fiscal_year):
    pillar = compute_pillar("esg", fiscal_year)

Stack Trace:
  File "/home/aydev/.local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_plan.py", line 299, in dagster_event_sequence_for_step
    raise dagster_user_error.user_exception
  File "/home/aydev/.local/lib/python3.10/site-packages/dagster/_core/execution/plan/utils.py", line 47, in solid_execution_error_boundary
    yield
  File "/home/aydev/.local/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 430, in iterate_with_context
    next_output = next(iterator)
  File "/home/aydev/.local/lib/python3.10/site-packages/dagster/_core/execution/plan/execute_step.py", line 557, in _gen_fn
    gen_output = output_manager.handle_output(output_context, output.value)
  File "/home/aydev/.local/lib/python3.10/site-packages/dagster/_core/storage/fs_io_manager.py", line 176, in handle_output
    pickle.dump(obj, write_obj, PICKLE_PROTOCOL)
  File "stringsource", line 2, in _catboost._PoolBase.__reduce_cython__

Hello all, getting above issue on

py31

with

dagster

latest (

1.0.15

). Seems like an issue with pickling C code, not sure how to about that. Any tips? :]

I’m running into some problems with a db resource. I’m starting to wonder if the connection is opening and closing before queries are run or if the lifecycle is different than what I would expect. Thus, I’m looking for some clarification on resource life cycles in dagster. Is the resource set up for each use in an op? At the start end of an op? Only once for a job/graph? Note I have tried to RTFM here: However, it’s referencing processes and steps which aren’t native dagster terms (as best I can tell).

#############
# Sample: 
#############

@resource 
@contextmanager 
def db_connection(): 
	try: 
		conn = psycopg2.connect()
        yield conn
	finally: 
		conn.close() 

@op(required_resource_keys={'db_connection'}
def my_op(context): 
	conn = context.resources.db_connection
	data = pd.read_sql("select * from foo", conn)
    data2 = pd.read_sql("select * from bar", conn)
    …

@op(required_resource_keys={'db_connection'}
def another_op(context): 
	conn = context.resources.db_connection
	data = pd.read_sql("select * from hello_world", conn)
    …

Questions: • my_op: Will the conn be created once and used twice? • my_op: Will the conn be created once, used once, and then fail on data2? • another_op: I’d expect to be able to reuse the resource, but it’s almost like the connection is being closed after my_op is run and can’t be reused. • General: Could the connection open and close on init? Or will the context manager only hit the finally once it's been used?

Hello, I am currently using this to get an environment variable

{
  "env": "ID_NUMBER"
}

However, it is returned as a string. Is there a way to read this as an integer? My env file has this

export ID_NUMBER=5

Hello, how do I go about setting "host" to the AWS Service Discovery for the gRPC server? workspace.yaml

load_from:
 - grpc_server:
    host:  <http://my_service_discovery.name.space|my_service_discovery.name.space>
    port: 4266
    location_name: "grpc_server"

Is the above acceptable or do I need to grab the IP address from the Discovery Service then feed it into "workspace.yaml"? Edit: Turns out you can just use the namespace as host. For anyone struggling to access your Dagster site and reload the repo, you have to add your gRPC port and the HTTP port (80) to the security group associated with your Dagster setup. AWS has a Route Analyzer that you can use to see if the daemon can reach the gRPC server. That helped me the most.

Hi, i am looking for an example of how to set up a source asset of a CSV file on AWS s3. I want to use the source asset in a downstream asset definition, so it is only downloaded during asset materialisation.

Hey, is it possible to "re-run from failure" but change the config slightly before the rerun?

Hi, I have been trying to get dagster setup with an M1 mac for the last 3 days and have had no success so far. Initially I was getting an error installing grpcio 1.43.0 with python 3.11 and with python3.10 grpcio 1.43.0 installs but when I try to run dagit I get the below error. Any ideas?

ImportError: dlopen [...] lib/python3.10/site-packages/grpc/_cython/cygrpc.cpython-310-darwin.so' (mach-o file, but is an incompatible architecture (have (x86_64), need (arm64e)))