Subflows
Subflows let you build modular and reusable workflow components.
They work like function calls: executing a subflow creates a new flow run from within another flow.
Why use a subflow?
Subflows allow you to build modular and reusable components that you can use across multiple flows. For example, you might define a subflow that handles error alerts by posting to Slack and email. By using a Subflow, you can reuse these two tasks together for all flows that you want to send error notifications, instead of having to copy the individual tasks for every flow.
How to declare a subflow
To call a flow from another flow, use the io.kestra.plugin.core.flow.Subflow
task, and in that task, specify the flowId
and namespace
of the subflow that you want to execute. You can also specify custom inputs
, similar to passing arguments to a function.
The optional properties wait
and transmitFailed
control the execution behavior. By default, if wait
is omitted or set to false
, the parent flow continues without waiting for the subflow to finish. The transmitFailed
property determines whether a failure in the subflow execution should cause the parent flow to fail.
Practical example
A subflow can encapsulate critical business logic, making it reusable across flows and easier to test in isolation.
Here is a simple example of a subflow:
id: critical_service
namespace: company.team
tasks:
- id: return_data
type: io.kestra.plugin.jdbc.duckdb.Query
sql: |
INSTALL httpfs;
LOAD httpfs;
SELECT sum(total) as total, avg(quantity) as avg_quantity
FROM read_csv_auto('https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv', header=True);
store: true
outputs:
- id: some_output
type: STRING
value: "{{ outputs.return_data.uri }}"
In this example, return_data
outputs uri
of the query output. That URI is a reference to the internal storage location of the stored file. This output can be used in the parent flow to perform further processing.
id: parent_service
namespace: company.team
tasks:
- id: subflow_call
type: io.kestra.plugin.core.flow.Subflow
namespace: company.team
flowId: critical_service
wait: true
transmitFailed: true
- id: log_subflow_output
type: io.kestra.plugin.scripts.shell.Commands
taskRunner:
type: io.kestra.plugin.core.runner.Process
commands:
- cat "{{ outputs.subflow_call.outputs.some_output }}"
The outputs
map task IDs to their results. Here, the parent flow accesses the some_output
value from the subflow_call
task.
Subflow properties
Below is a full list of all properties of the io.kestra.plugin.core.flow.Subflow
task. You don’t need to memorize all properties — the task documentation always lists them.
Field | Description |
---|---|
flowId | The subflow's identifier |
namespace | The namespace where the subflow is located |
inheritLabels | Determines if the subflow inherits labels from the parent (default: false). |
inputs | Inputs passed to the subflow |
labels | Labels assigned to the subflow |
outputs (deprecated) | Allows passing outputs from the subflow execution to the parent flow. |
revision | The subflow revision to execute (defaults to the latest) |
scheduleDate | Schedule subflow execution on a specific date rather than immediately. |
transmitFailed | If true, parent flow fails on subflow failure (requires wait to be true). |
wait | If true, parent flow waits for subflow completion (default: true). |
Passing data between parent and child flows
Flows can emit outputs that can be accessed by the parent flow. Using the io.kestra.plugin.core.flow.Subflow
task you can call any flow as a subflow and access its outputs in downstream tasks. For more details and examples, check the Outputs page.
Accessing Outputs from a subflow execution
Outputs include the execution ID, extracted outputs, and the final state (if wait
is true).
To sum up, subflows improve maintainability of complex workflows. They allow you to build modular and reusable workflow components and share them across multiple namespaces, projects, and teams.
Here’s an example of a subflow with explicitly defined outputs.
id: flow_outputs
namespace: company.team
tasks:
- id: mytask
type: io.kestra.plugin.core.debug.Return
format: this is a task output used as a final flow output
outputs:
- id: final
type: STRING
value: "{{ outputs.mytask.value }}"
We can access these outputs from a parent task as seen in the example below:
id: parent_flow
namespace: company.team
tasks:
- id: subflow
type: io.kestra.plugin.core.flow.Subflow
flowId: flow_outputs
namespace: company.team
wait: true
- id: log_subflow_output
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.subflow.outputs.final }}"
For more details, see the sublow outputs documentation.
Passing inputs to a subflow
You can pass inputs to a Subflow task. The example below passes two inputs to a subflow.
Subflow:
id: subflow_example
namespace: company.team
inputs:
- id: http_uri
type: STRING
tasks:
- id: download
type: io.kestra.plugin.core.http.Request
uri: "{{ inputs.http_uri }}"
- id: log
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.download.body }}"
outputs:
- id: data
type: STRING
value: "{{ outputs.download.body }}"
Parent flow:
id: inputs_subflow
namespace: company.team
inputs:
- id: url
type: STRING
tasks:
- id: subflow
type: io.kestra.plugin.core.flow.Subflow
flowId: subflow_example
namespace: company.team
inputs:
http_uri: "{{ inputs.url }}"
wait: true
- id: hello
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.subflow.outputs.data }}"
In this example, the parent flow successfully passes an input to the subflow.
Nested inputs
In the example below, the flow extracts JSON data from a REST API and passes it to a subflow as a nested input:
id: extract_json
namespace: company.team
tasks:
- id: api
type: io.kestra.plugin.core.http.Request
uri: https://dummyjson.com/users
- id: read_json
type: io.kestra.plugin.core.log.Log
message: "{{ outputs.api.body }}"
- id: subflow
type: io.kestra.plugin.core.flow.Subflow
namespace: company.team
flowId: subflow
inputs:
users.firstName: "{{ outputs.api.body | jq('.users') | first | first | jq('.firstName') | first }}"
users.lastName: "{{ outputs.api.body | jq('.users') | first | first | jq('.lastName') | first }}"
wait: true
transmitFailed: true
To provide type validation to extracted JSON fields, you can use nested inputs in the subflow definition:
id: subflow
namespace: company.team
inputs:
- id: users.firstName
type: STRING
defaults: Rick
- id: users.lastName
type: STRING
defaults: Astley
tasks:
- id: process_user_data
type: io.kestra.plugin.core.log.Log
message: hello {{ inputs.users }}
You can then pass the entire users
object, including nested fields, to any task in the subflow.
Was this page helpful?