https://dagster.io/ logo
#ask-ai
Title
# ask-ai
a

Akira Renbokoji

03/04/2024, 4:43 PM
I have an
@op
with
Out()
that does
return a, b
how can I change it so that it is a generator instead
yield a, b
s

Scout

03/04/2024, 4:43 PM
In Dagster, when you have an
@op
that needs to yield multiple outputs, you should yield
Output
objects separately, each with a specified
output_name
. The
output_name
corresponds to the names you define in the
out
parameter of the
@op
decorator. Here's how you can modify your
@op
to use a generator that yields multiple outputs:
Copy code
python from dagster import Output, op, Out  @op(out={"output_a": Out(), "output_b": Out()}) def my_op():     # Compute or retrieve values for a and b     a = compute_a()     b = compute_b()          yield Output(a, output_name="output_a")     yield Output(b, output_name="output_b")
In this example,
compute_a()
and
compute_b()
are placeholder functions that represent the computation or retrieval of the values for
a
and
b
. The
Output
objects are yielded with their respective
output_name
that matches the keys defined in the
out
parameter. Remember that when you yield outputs from an
@op
, you cannot use type annotations for the return type as you would with a regular Python function that returns a tuple. Instead, you specify the output types and names using the
out
parameter of the
@op
decorator, and then yield
Output
objects accordingly.
a

Akira Renbokoji

03/04/2024, 4:50 PM
@Scout It looks like the Dagster
@op
generator yields all the values at once instead of each time its called. How do you recommend this be passed to another
@op
with
In()
?
s

Scout

03/04/2024, 4:50 PM
In Dagster, when you have an
@op
that yields multiple outputs, you can pass these outputs to another
@op
by connecting them in a graph or job. Each output from the yielding
@op
can be individually passed to downstream ops by referencing the specific output names. Here's an example of how you can pass the yielded outputs to another `@op`:
Copy code
python from dagster import op, Out, Output, In, job  # Define the op that yields multiple outputs @op(out={"output_a": Out(), "output_b": Out()}) def my_yielding_op():     a = compute_a()  # Replace with actual computation     b = compute_b()  # Replace with actual computation     yield Output(a, output_name="output_a")     yield Output(b, output_name="output_b")  # Define the op that takes multiple inputs @op def my_receiving_op(input_a, input_b):     # Do something with input_a and input_b     process_inputs(input_a, input_b)  # Define a job that connects the ops @job def my_job():     output_a, output_b = my_yielding_op()     my_receiving_op(input_a=output_a, input_b=output_b)
In this example,
my_yielding_op
yields two outputs,
output_a
and
output_b
. In the
my_job
definition, these outputs are captured as
output_a
and
output_b
and then passed as arguments to
my_receiving_op
using the parameter names
input_a
and
input_b
. If you want to pass the outputs to an
@op
that explicitly expects inputs with
In()
, you can define the receiving
@op
like this:
Copy code
python # Define the op that takes multiple inputs with explicit In() annotations @op(ins={"input_a": In(), "input_b": In()}) def my_receiving_op_explicit(input_a, input_b):     # Do something with input_a and input_b     process_inputs(input_a, input_b)
The
In()
annotations are not strictly necessary if you're just passing values, but they can be used to provide additional metadata or configuration for the inputs. Remember to replace
compute_a
,
compute_b
, and
process_inputs
with your actual computation and processing logic. The
my_job
function connects the ops together, ensuring that the outputs from
my_yielding_op
are passed to
my_receiving_op
.
a

Akira Renbokoji

03/04/2024, 4:56 PM
@Scout So if I have a generator
foo()
I only want it to return data when I call it.
Copy code
@op(out={"foo_first": foo_first}, out={"foo_second": foo_second)
def foo():
    for foo_first, foo_second in some_data:
        yield Output(foo_first, output_name="foo_first")
        yield Output(foo_second, output_name="foo_second")

@op(ins={"foo_first": In(), "foo_second": In()):
def bar(foo_first, foo_second):
    do_stuff(foo_first, foo_second)

@graph
def try_this():
    foo = foo()
    bar(foo)
    baz(foo)
Is something like this possible? If not, how should I go about it? Is this possible?
s

Scout

03/04/2024, 4:57 PM
In Dagster, when you define an
@op
that yields multiple outputs, you should yield each output separately within a single iteration of the loop, not in separate yields, because each yield statement is considered as a separate event in the execution of the op. Additionally, the syntax you've used for defining multiple outputs in the
@op
decorator is incorrect. Here's how you can correctly define an
@op
that yields multiple outputs:
Copy code
python from dagster import op, Out, Output  @op(out={"foo_first": Out(), "foo_second": Out()}) def foo():     for foo_first, foo_second in some_data:         yield Output(foo_first, output_name="foo_first")         yield Output(foo_second, output_name="foo_second")
However, the above code will not work as intended because it yields
foo_first
and
foo_second
in separate iterations, which is not the correct way to yield multiple outputs in Dagster. Instead, you should yield all outputs for a single iteration together, like this:
Copy code
python from dagster import op, Out, Output  @op(out={"foo_first": Out(), "foo_second": Out()}) def foo():     for foo_first, foo_second in some_data:         yield Output(foo_first, output_name="foo_first")         yield Output(foo_second, output_name="foo_second")
Now, to connect this op to downstream ops in a graph, you would do the following:
Copy code
python from dagster import graph  @op def bar(foo_first, foo_second):     do_stuff(foo_first, foo_second)  @op def baz(foo_first, foo_second):     do_other_stuff(foo_first, foo_second)  @graph def try_this():     foo_results = foo()     bar(foo_first=foo_results.foo_first, foo_second=foo_results.foo_second)     baz(foo_first=foo_results.foo_first, foo_second=foo_results.foo_second)
In this graph,
foo_results
is an object that contains both outputs from the
foo
op. You can then pass these outputs to
bar
and
baz
by specifying the output names (
foo_first
and
foo_second
). Please note that the
do_stuff
and
do_other_stuff
functions should be replaced with your actual logic, and
some_data
should be replaced with the actual data you're iterating over.
a

Akira Renbokoji

03/04/2024, 4:58 PM
@Scout So if I have a generator
foo()
I only want it to return data when I call it.
Copy code
@op(out={"foo_first": foo_first}, out={"foo_second": foo_second)
def foo():
    for foo_first, foo_second in some_data:
        yield Output(foo_first, output_name="foo_first")
        yield Output(foo_second, output_name="foo_second")

@op(ins={"foo_first": In(), "foo_second": In()):
def bar(foo_first, foo_second):
    do_stuff(foo_first, foo_second)

@graph
def try_this():
    bar(foo())
    baz(foo())
Is something like this possible? If not, how should I go about it? Is this possible?
s

Scout

03/04/2024, 4:58 PM
In Dagster, the way you've written the
@op
decorator with multiple
out
parameters is not correct. You should define a single
out
parameter with a dictionary specifying each output. Additionally, the
@graph
definition is not correctly connecting the outputs of
foo
to the inputs of
bar
and
baz
. Here's how you can correct these issues: First, define the
foo
op with multiple outputs correctly:
Copy code
python from dagster import op, Out, Output  @op(out={"foo_first": Out(), "foo_second": Out()}) def foo():     for foo_first_value, foo_second_value in some_data:         yield Output(foo_first_value, output_name="foo_first")         yield Output(foo_second_value, output_name="foo_second")
Then, define the
bar
and
baz
ops that take inputs:
Copy code
python @op def bar(foo_first, foo_second):     do_stuff(foo_first, foo_second)  @op def baz(foo_first, foo_second):     do_other_stuff(foo_first, foo_second)
Finally, define the graph that connects the outputs of
foo
to the inputs of
bar
and
baz
. However, you cannot call
foo()
directly in the
bar
and
baz
invocations because
foo()
is an op that yields multiple outputs. Instead, you should capture the outputs of
foo
and pass them to
bar
and `baz`:
Copy code
python from dagster import graph  @graph def try_this():     foo_results = foo()     bar(foo_first=foo_results.foo_first, foo_second=foo_results.foo_second)     baz(foo_first=foo_results.foo_first, foo_second=foo_results.foo_second)
In this graph,
foo_results
is a structure that contains both outputs from the
foo
op, and you can access each output by its name (
foo_first
and
foo_second
). These named outputs are then passed as inputs to
bar
and
baz
. Please note that
some_data
should be replaced with the actual data you're iterating over, and
do_stuff
and
do_other_stuff
should be replaced with your actual logic for
bar
and
baz
, respectively. If you want
foo
to act as a generator that only computes values when called, you would typically structure your code as shown above. Each time
foo
is invoked within a Dagster job or graph, it will execute and yield its outputs, which can then be passed to other ops. Dagster handles the execution flow, ensuring that
foo
is called and its outputs are passed to downstream ops as needed.
a

Akira Renbokoji

03/04/2024, 5:03 PM
@Scout I'm getting the error: foo() returned foo_first multiple times
s

Scout

03/04/2024, 5:03 PM
The error message you're encountering suggests that the
foo
op is yielding outputs with the same output name multiple times, which is not allowed in Dagster. Each output name must be unique within the context of a single op execution. Here's how you can modify the
foo
op to yield multiple outputs correctly:
Copy code
python from dagster import op, Out, Output  @op(out={"foo_first": Out(), "foo_second": Out()}) def foo():     # Assuming some_data is an iterable of tuples (foo_first_value, foo_second_value)     for foo_first_value, foo_second_value in some_data:         yield Output(foo_first_value, output_name="foo_first")         yield Output(foo_second_value, output_name="foo_second")
However, the above code will yield the
foo_first
and
foo_second
outputs in separate iterations, which is not the correct way to yield multiple outputs in Dagster. Instead, you should yield all outputs for a single iteration together, like this:
Copy code
python from dagster import op, Out, Output  @op(out={"foo_first": Out(), "foo_second": Out()}) def foo():     # Assuming some_data is an iterable of tuples (foo_first_value, foo_second_value)     for foo_first_value, foo_second_value in some_data:         yield Output(foo_first_value, output_name="foo_first")         yield Output(foo_second_value, output_name="foo_second")         # Break after yielding both outputs to ensure we don't yield the same output names again         break
In this corrected version,
foo
yields both
foo_first
and
foo_second
once per iteration and then breaks out of the loop to prevent yielding outputs with the same names again. If you intend for
foo
to yield multiple pairs of outputs (one pair per iteration), you should consider restructuring your graph or job to handle this correctly. Dagster expects each output name to be yielded exactly once per op execution. If you need further assistance, please provide more context or clarify your use case, and I can offer more specific guidance.
a

Akira Renbokoji

03/04/2024, 5:13 PM
@Scout I'm still getting the error: Compute for op "foo()" returned "foo_first" multiple times.
s

Scout

03/04/2024, 5:13 PM
The error you're encountering indicates that the
foo
op is yielding the same output name more than once during its execution, which is not allowed. In Dagster, each output name must be yielded exactly once. To fix this, you should ensure that each output name is yielded only once per execution of the op. Here's an example of how you can do this:
Copy code
python
from dagster import op, Out, Output

@op(out={"foo_first": Out(), "foo_second": Out()})
def foo():
    # Assuming some_data is an iterable of tuples (foo_first_value, foo_second_value)
    for foo_first_value, foo_second_value in some_data:
        # Yield both outputs at once, then break to ensure we don't yield the same output names again
        yield Output(foo_first_value, output_name="foo_first")
        yield Output(foo_second_value, output_name="foo_second")
        break  # Important: break after yielding both outputs to prevent yielding the same output names again
However, if
some_data
contains multiple pairs of data and you want to process each pair in a separate execution of
foo
, you should not use a loop inside the
foo
op. Instead, you should structure your graph or job to call
foo
multiple times, once for each pair of data. If you want to process each pair of data from
some_data
and pass them to other ops like
bar
and
baz
, you might need to restructure your graph to iterate over
some_data
outside of the
foo
op. Here's an example of how you might do this:
Copy code
python
from dagster import graph, op, Out, Output

# Define the op that processes a single pair of data
@op(out={"foo_first": Out(), "foo_second": Out()})
def foo(data_pair):
    foo_first_value, foo_second_value = data_pair
    yield Output(foo_first_value, output_name="foo_first")
    yield Output(foo_second_value, output_name="foo_second")

# Define the ops that consume the outputs of foo
@op
def bar(foo_first, foo_second):
    # Process the outputs
    pass

@op
def baz(foo_first, foo_second):
    # Process the outputs
    pass

# Define a graph that processes each pair of data
@graph
def process_data_pairs():
    for data_pair in some_data:
        foo_results = foo(data_pair)
        bar(foo_results.foo_first, foo_results.foo_second)
        baz(foo_results.foo_first, foo_results.foo_second)

# Now you can execute the graph for each pair of data
process_data_pairs()
In this example,
some_data
should be a list or iterable of tuples, where each tuple represents a pair of data to be processed. The
process_data_pairs
graph calls
foo
for each pair, then passes the results to
bar
and
baz
. Please note that this example assumes that
some_data
is available at graph construction time. If
some_data
is not available until runtime, you would need to handle this differently, potentially by us
ing dynamic orchestration with
DynamicOut
and
DynamicOutput
. If you continue to encounter issues, please provide additional context about how
some_data
is generated and how you intend to process it within your Dagster job or graph.
a

Akira Renbokoji

03/04/2024, 5:16 PM
@Scout show me this example with DynamicOut and DynamicOutput
s

Scout

03/04/2024, 5:16 PM
To handle a scenario where you have an iterable of data that you want to process dynamically at runtime, you can use
DynamicOut
and
DynamicOutput
in Dagster. This allows you to dynamically generate downstream ops for each piece of data. Here's an example of how you can use
DynamicOut
and
DynamicOutput
to process each pair of data from `some_data`:
Copy code
python from dagster import op, DynamicOut, DynamicOutput, graph, job  # Define the op that dynamically yields multiple pairs of data @op(out=DynamicOut()) def foo():     # Assuming some_data is an iterable of tuples (foo_first_value, foo_second_value)     for idx, (foo_first_value, foo_second_value) in enumerate(some_data):         yield DynamicOutput((foo_first_value, foo_second_value), mapping_key=str(idx))  # Define the ops that consume the outputs of foo @op def bar(data_pair):     foo_first, foo_second = data_pair     # Process the outputs     pass  @op def baz(data_pair):     foo_first, foo_second = data_pair     # Process the outputs     pass  # Define a job that processes each pair of data dynamically @job def process_data_pairs_job():     dynamic_results = foo()     # Map each dynamic output to the bar and baz ops     dynamic_results.map(bar)     dynamic_results.map(baz)  # Now you can execute the job process_data_pairs_job()
In this example,
foo
is an op that yields a
DynamicOutput
for each pair of data in
some_data
. Each
DynamicOutput
contains the data pair and a unique
mapping_key
. The
dynamic_results
object represents the dynamic output from
foo
. The
dynamic_results.map(bar)
and
dynamic_results.map(baz)
lines map each dynamic output to the
bar
and
baz
ops, respectively. This means that for each pair of data yielded by
foo
, a separate instance of
bar
and
baz
will be executed. Please note that
some_data
should be replaced with the actual data you're iterating over. This example assumes that
some_data
is available at runtime and that
bar
and
baz
are designed to process a tuple containing a pair of data. The
DynamicOut
and
DynamicOutput
constructs are particularly useful when the exact number of data pairs to process is not known until runtime, allowing for dynamic parallel execution of downstream ops.
a

Akira Renbokoji

03/04/2024, 5:19 PM
@Scout Let's start again from scratch. I like the Dynamic Out and Output stuff. Say I have a list of key, value dictionary pairs. How would I go about setting up a graph that calls ops with Dynamic Out and Output? I don't think I need to use the keyword yield.
s

Scout

03/04/2024, 5:19 PM
Certainly! If you have a list of key-value dictionary pairs and you want to process each item dynamically, you can set up a Dagster graph that uses
DynamicOut
and
DynamicOutput
without using the
yield
keyword. Instead, you can return a list of
DynamicOutput
objects from your op. Here's an example of how you might set up such a graph:
Copy code
python from dagster import op, DynamicOut, DynamicOutput, graph, job  # Assume you have a list of dictionary pairs like this: list_of_dict_pairs = [     {"key": "key1", "value": "value1"},     {"key": "key2", "value": "value2"},     # ... more pairs ]  # Define the op that processes the list of dictionary pairs and returns DynamicOutputs @op(out=DynamicOut()) def process_dict_pairs():     outputs = []     for idx, dict_pair in enumerate(list_of_dict_pairs):         # Create a DynamicOutput for each dictionary pair         outputs.append(DynamicOutput(value=dict_pair, mapping_key=str(idx)))     return outputs  # Define the op that consumes each dictionary pair @op def process_pair(dict_pair):     # Process the dictionary pair     # For example, you might want to do something with the key and value     key = dict_pair["key"]     value = dict_pair["value"]     # ... your processing logic here  # Define a graph that sets up the dynamic processing @graph def process_pairs_graph():     dynamic_pairs = process_dict_pairs()     # Map each dynamic output to the process_pair op     dynamic_pairs.map(process_pair)  # Define a job that encapsulates the graph @job def process_pairs_job():     process_pairs_graph()  # Now you can execute the job process_pairs_job()
In this example,
process_dict_pairs
is an op that takes no inputs and processes a predefined list of dictionary pairs. For each dictionary pair, it creates a
DynamicOutput
object with a unique
mapping_key
. The op then returns a list of these
DynamicOutput
objects. The
process_pair
op is defined to take a single dictionary pair as input and contains the logic for processing that pair. The
process_pairs_graph
graph sets up the dynamic processing by mapping each
DynamicOutput
from
process_dict_pairs
to the
process_pair
op. Finally,
process_pairs_job
is a job that encapsulates the graph, and you can execute this job to run the dynamic processing. This setup allows Dagster to dynamically generate separate instances of
process_pair
for each item in
list_of_dict_pairs
at runtime.
a

Akira Renbokoji

03/04/2024, 5:20 PM
@Scout now can u do it again with the keyword yield?
s

Scout

03/04/2024, 5:20 PM
Certainly! If you prefer to use the
yield
keyword to emit
DynamicOutput
objects from an op, you can do so by yielding each
DynamicOutput
within a loop. Here's how you can set up a Dagster graph that processes each key-value dictionary pair using `yield`:
Copy code
python from dagster import op, DynamicOut, DynamicOutput, graph, job  # Assume you have a list of dictionary pairs like this: list_of_dict_pairs = [     {"key": "key1", "value": "value1"},     {"key": "key2", "value": "value2"},     # ... more pairs ]  # Define the op that processes the list of dictionary pairs and yields DynamicOutputs @op(out=DynamicOut()) def process_dict_pairs():     for idx, dict_pair in enumerate(list_of_dict_pairs):         # Yield a DynamicOutput for each dictionary pair         yield DynamicOutput(value=dict_pair, mapping_key=str(idx))  # Define the op that consumes each dictionary pair @op def process_pair(dict_pair):     # Process the dictionary pair     # For example, you might want to do something with the key and value     key = dict_pair["key"]     value = dict_pair["value"]     # ... your processing logic here  # Define a graph that sets up the dynamic processing @graph def process_pairs_graph():     dynamic_pairs = process_dict_pairs()     # Map each dynamic output to the process_pair op     dynamic_pairs.map(process_pair)  # Define a job that encapsulates the graph @job def process_pairs_job():     process_pairs_graph()  # Now you can execute the job process_pairs_job()
In this example, the
process_dict_pairs
op uses a loop to iterate over
list_of_dict_pairs
, and for each dictionary pair, it yields a
DynamicOutput
object with a unique
mapping_key
. The rest of the graph and job definitions remain the same as in the previous example. By using
yield
, you instruct Dagster to emit each
DynamicOutput
as soon as it's generated within the loop. This can be particularly useful if you have a large number of items to process or if the processing of each item is independent and can be parallelized. The
process_pair
op is defined to take a single dictionary pair as input and contains the logic for processing that pair. The
process_pairs_graph
graph sets up the dynamic processing by mapping each
DynamicOutput
from
process_dict_pairs
to the
process_pair
op. Finally,
process_pairs_job
is a job that encapsulates the graph, and you can execute this job to run the dynamic processing.
a

Akira Renbokoji

03/04/2024, 5:33 PM
@Scout So will
dynamic_pairs.map(process_pair)
be called numerous times for each unique pair in the list?
s

Scout

03/04/2024, 5:33 PM
Yes, when you use
dynamic_pairs.map(process_pair)
in the graph, Dagster will create a separate instance of the
process_pair
op for each
DynamicOutput
emitted by the
process_dict_pairs
op. Each instance of
process_pair
will be called with a unique dictionary pair from the list. Here's what happens step by step: 1. The
process_dict_pairs
op iterates over
list_of_dict_pairs
and yields a
DynamicOutput
for each dictionary pair, each with a unique
mapping_key
. 2. The
dynamic_pairs
object in the graph represents the collection of dynamic outputs from
process_dict_pairs
. 3. When
dynamic_pairs.map(process_pair)
is called, Dagster maps the
process_pair
op over each item in
dynamic_pairs
. This means that for each
DynamicOutput
yielded by
process_dict_pairs
, a new instance of
process_pair
will be invoked with the value contained in that
DynamicOutput
. 4. Each invocation of
process_pair
is independent and can potentially be executed in parallel, depending on the executor configuration and the resources available. 5. As a result,
process_pair
will be called as many times as there are items in
list_of_dict_pairs
, with each call receiving a different dictionary pair as input. This dynamic mapping allows for parallel processing of the items in the list, which can be beneficial for performance, especially when dealing with a large number of items or when the processing of each item is computationally expensive.
a

Akira Renbokoji

03/04/2024, 5:48 PM
@Scout I'm getting a "required missing config at ops" error for the DynamicOutput
s

Scout

03/04/2024, 5:48 PM
The "required missing config at ops" error typically indicates that Dagster expects some configuration for an op within the job, but it hasn't been provided. This can happen if an op has a
config_schema
defined and the necessary configuration is not supplied when the job is executed. When using
DynamicOutput
and
DynamicOut
, the error should not be related to the dynamic nature of the outputs themselves, but rather to the configuration of the ops that are being dynamically mapped. Here's a general approach to troubleshoot and resolve this issue: 1. Check Config Schema: Ensure that all ops that require configuration have a
config_schema
defined and that the schema is correct. 2. Provide Required Config: When executing the job, make sure to provide the required configuration for all ops. This can be done in a YAML file or directly in the code when using
execute_in_process
or similar methods. 3. Review Job Execution: If you're executing the job from the command line or the Dagster UI, ensure that the configuration is being passed correctly. For command-line execution, you would typically use the
--config
flag followed by the path to your YAML configuration file. 4. Check for Typos: A common issue is typos in the configuration keys or in the op names. Make sure that the keys in your configuration match the
config_schema
and that the op names match those used in your job definition. 5. Validate Configuration: You can use Dagster's tools to validate your configuration before running the job. This can help catch errors in the configuration structure or values. If you continue to face issues, please provide more details about your job definition, the ops that require configuration, and how you are providing the configuration when executing the job. This information will help diagnose the problem more accurately.
a

Akira Renbokoji

03/04/2024, 5:49 PM
@Scout For the DynamicOutput example you showed did any of them need a config?
s

Scout

03/04/2024, 5:49 PM
In the DynamicOutput example I provided earlier, there was no explicit configuration required for the ops. The ops were defined without a
config_schema
, which means they do not expect any run-time configuration to be provided. Here's a recap of the example with the
DynamicOut
and `DynamicOutput`:
Copy code
python from dagster import op, DynamicOut, DynamicOutput, graph, job  # Assume you have a list of dictionary pairs like this: list_of_dict_pairs = [     {"key": "key1", "value": "value1"},     {"key": "key2", "value": "value2"},     # ... more pairs ]  # Define the op that processes the list of dictionary pairs and yields DynamicOutputs @op(out=DynamicOut()) def process_dict_pairs():     for idx, dict_pair in enumerate(list_of_dict_pairs):         # Yield a DynamicOutput for each dictionary pair         yield DynamicOutput(value=dict_pair, mapping_key=str(idx))  # Define the op that consumes each dictionary pair @op def process_pair(dict_pair):     # Process the dictionary pair     # For example, you might want to do something with the key and value     key = dict_pair["key"]     value = dict_pair["value"]     # ... your processing logic here  # Define a graph that sets up the dynamic processing @graph def process_pairs_graph():     dynamic_pairs = process_dict_pairs()     # Map each dynamic output to the process_pair op     dynamic_pairs.map(process_pair)  # Define a job that encapsulates the graph @job def process_pairs_job():     process_pairs_graph()  # Now you can execute the job process_pairs_job()
In this example, neither
process_dict_pairs
nor
process_pair
requires configuration. If you encounter a "required missing config at ops" error, it suggests that there might be another op within your job or graph that does require configuration, or there might be a misunderstanding about how the configuration is being provided. If you have ops that do require configuration, you would need to provide that configuration when you execute the job. The documentation provides guidance on how to supply run-time configuration for jobs, which can be done via YAML files, environment variables, or directly in code when using methods like
execute_in_process
. If you need to add configuration to the ops in the example, you would define a
config_schema
for each op that requires it and then ensure that the necessary configuration is provided at run time. If you're still having trouble, please share more details about your specific use case, and I can provide more targeted assistance.
15 Views