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

Brandon Peebles

03/05/2024, 9:53 PM
How can I create a job that conditionally runs an op depending on whether it's a branch deployment or not? Assume I have
op_one
that is a dependency of
op_two
such that I use the output of
op_one
in
op_two
. I want to skip running
op_one
in branch deployments since it is not needed
s

Scout

03/05/2024, 9:53 PM
To create a job that conditionally runs an op depending on whether it's a branch deployment or not, you can determine the current deployment type at runtime and use that information to conditionally execute your ops. The documentation provides an example of how to determine if the current deployment is a branch deployment using an environment variable:
Copy code
python import os  def get_current_env():     is_branch_depl = os.getenv("DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT") == "1"     assert is_branch_depl is not None  # env var must be set     return "branch" if is_branch_depl else "prod"
In this function, the
DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT
environment variable is checked, and if it is equal to
"1"
, it indicates that the current deployment is a branch deployment. Using this information, you can write a job that includes logic to conditionally execute
op_one
based on the deployment type. However, since the documentation does not provide a direct example of how to conditionally skip an op within a job, I will provide general guidance based on the principles of Dagster. You would typically define your job with both
op_one
and
op_two
, where
op_two
depends on
op_one
. To conditionally skip
op_one
, you could use a pattern where
op_one
always runs but performs no action or returns a default value in branch deployments. Then,
op_two
can use the output of
op_one
as needed. Here's a conceptual example:
Copy code
python from dagster import job, op  @op def op_one(context):     if get_current_env() == "branch":         <http://context.log.info|context.log.info>("Skipping op_one in branch deployment.")         # Return a default value or perform no action         return None     else:         # Perform the actual computation or action         return compute_something()  @op def op_two(context, input_from_op_one):     # Use the output from op_one as needed     if input_from_op_one is not None:         # Perform actions with the input         pass     else:         <http://context.log.info|context.log.info>("Received no input from op_one, proceeding with alternative logic.")  @job def conditional_job():     op_two(op_one())
In this example,
op_one
checks the deployment type and decides whether to perform its computation.
op_two
takes the output of
op_one
and handles the case where
op_one
did not perform its usual computation. Please note that this is a conceptual example to illustrate the approach. You would need to adapt this pattern to your specific use case and ensure that
op_two
can handle the case where
op_one
does not provide its usual output.
b

Brandon Peebles

03/05/2024, 9:55 PM
@Scout no I want to skip op_one entirely. I want the conditional logic to live inside the
job
definition. I'd rather not execute an op and use a credit if it's not needed.
s

Scout

03/05/2024, 9:55 PM
To skip
op_one
entirely based on a condition within the
job
definition, you can use dynamic orchestration with Dagster's
DynamicOut
and
DynamicOutput
to conditionally yield outputs that determine the execution graph at runtime. This allows you to create a job where the execution of
op_two
does not depend on
op_one
when certain conditions are met, such as a branch deployment. Here's a conceptual example using dynamic orchestration to skip `op_one`:
Copy code
python from dagster import job, op, DynamicOut, DynamicOutput, Out  @op(out=DynamicOut()) def branch_check_op():     if get_current_env() == "branch":         # Skip op_one in branch deployment         yield DynamicOutput(value=None, mapping_key="skip_op_one")     else:         # Proceed with op_one in other deployments         yield DynamicOutput(value=None, mapping_key="run_op_one")  @op def op_one():     # Perform the actual computation or action     return compute_something()  @op def op_two(input_from_op_one=None):     # Use the output from op_one as needed     # Perform actions with the input     pass  @job def conditional_job():     branch_decision = branch_check_op()          # Only run op_one if the branch_check_op yields the "run_op_one" output     op_one_result = branch_decision.map(op_one).collect()          # Pass the result of op_one to op_two, which can handle None input     op_two(op_one_result)
In this example,
branch_check_op
is an op that dynamically determines whether to proceed with
op_one
based on the deployment environment. It yields a
DynamicOutput
with a mapping key that indicates whether to skip or run
op_one
. The
map
method is then used to conditionally apply
op_one
to the output of
branch_check_op
. The
collect
method is used to gather the dynamic outputs into a single output that can be passed to
op_two
. Please note that this is a conceptual example to illustrate the approach. You would need to adapt this pattern to your specific use case and ensure that
op_two
can handle the case where it receives no input from
op_one
. Additionally, you should ensure that the dynamic orchestration features used in this example align with the current capabilities and best practices of the Dagster version you are using.
b

Brandon Peebles

03/05/2024, 9:57 PM
@Scout is it not possible to simply pass a custom
output
to the downstream op instead of one that was actually generated by an op?
s

Scout

03/05/2024, 9:58 PM
In Dagster, you can indeed pass a custom output to a downstream op instead of one that was actually generated by an upstream op. This can be done by using the
Out
and
Output
classes to explicitly yield outputs from ops, which can then be passed to downstream ops. This allows you to bypass the execution of an upstream op and provide a predetermined value directly to a downstream op. Here's an example of how you might structure your job to conditionally pass a custom output to `op_two`:
Copy code
python from dagster import job, op, Out, Output  @op(out={"condition": Out(), "result": Out(is_required=False)}) def check_condition_and_produce_result():     if get_current_env() == "branch":         # In a branch deployment, we skip op_one and provide a custom output         yield Output(True, "condition")         # Since op_one is skipped, we do not yield a result for it     else:         # In other deployments, we run op_one and provide its result         yield Output(False, "condition")         yield Output(compute_something(), "result")  @op def op_two(input_from_op_one=None):     # Use the output from op_one or the custom input as needed     # Perform actions with the input     pass  @job def conditional_job():     condition_result, op_one_result = check_condition_and_produce_result()     op_two(input_from_op_one=op_one_result)
In this example,
check_condition_and_produce_result
is an op that checks the deployment environment and yields two outputs: a condition flag and an optional result. If the condition indicates a branch deployment, it yields
True
for the condition and does not yield a result for
op_one
. If it's not a branch deployment, it yields
False
for the condition and the actual result of
op_one
. The
op_two
is then defined to accept an input that can either be the output of
op_one
or a custom value (in this case,
None
if
op_one
is skipped). The
conditional_job
job wires up these ops, and
op_two
will receive the appropriate input based on the condition. This approach allows you to control the flow of your job and the execution of ops based on runtime conditions without executing unnecessary ops.
14 Views