Control Execution Flow with Sequential, Parallel, and Loop Tasks

Flowable tasks control orchestration logic: branching, looping, and parallelizing work, without performing heavy computation.

Flowable tasks use expressions from the execution context to determine which tasks run next.

Sequential

Sequential runs child tasks one after another, with optional errors and finally hooks.

Tasks inside a Sequential block can reference sibling task outputs using {{ outputs.sibling_id.value }}.

id: sequential
namespace: company.team

tasks:
  - id: sequential
    type: io.kestra.plugin.core.flow.Sequential
    tasks:
      - id: first_task
        type: io.kestra.plugin.core.output.OutputValues
        values:
          result: "{{ task.id }} > {{ taskrun.startDate }}"

      - id: second_task
        type: io.kestra.plugin.core.log.Log
        message: "{{ task.id }} > {{ outputs.first_task.values.result }}"

  - id: last
    type: io.kestra.plugin.core.log.Log
    message: "{{ task.id }} > {{ taskrun.startDate }}"

For more details, check out the Sequential Task documentation.

Parallel

Parallel starts all child tasks concurrently, reducing total elapsed time when tasks are independent. Because branches run simultaneously, you cannot access the output of a sibling task from within the same Parallel block; outputs are only available to tasks that run after the Parallel task completes.

The concurrent property caps how many branches run at once (0 = no limit, the default).

id: parallel
namespace: company.team

tasks:
  - id: parallel
    type: io.kestra.plugin.core.flow.Parallel
    concurrent: 2
    tasks:
      - id: branch_1
        type: io.kestra.plugin.core.log.Log
        message: "{{ task.id }} > {{ taskrun.startDate }}"

      - id: branch_2
        type: io.kestra.plugin.core.log.Log
        message: "{{ task.id }} > {{ taskrun.id }}"

  - id: last
    type: io.kestra.plugin.core.log.Log
    message: "{{ task.id }} > {{ taskrun.startDate }}"

Sequential nested inside Parallel branches runs multi-step sequences concurrently:

  - id: parallel
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: sequence1
        type: io.kestra.plugin.core.flow.Sequential
        tasks:
          - id: task1
            type: io.kestra.plugin.core.log.Log
            message: "step 1a"
          - id: task2
            type: io.kestra.plugin.core.log.Log
            message: "step 1b"
      - id: sequence2
        type: io.kestra.plugin.core.flow.Sequential
        tasks:
          - id: task3
            type: io.kestra.plugin.core.log.Log
            message: "step 2a"
          - id: task4
            type: io.kestra.plugin.core.log.Log
            message: "step 2b"

For more task details, refer to the Parallel Task documentation.

Switch

Switch routes execution to a matching case based on the value of an expression. In the following example, an input determines which branch runs.

id: switch
namespace: company.team

inputs:
  - id: param
    type: BOOL

tasks:
  - id: decision
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.param }}"
    cases:
      true:
        - id: is_true
          type: io.kestra.plugin.core.log.Log
          message: "This is true"
      false:
        - id: is_false
          type: io.kestra.plugin.core.log.Log
          message: "This is false"

For more plugin details, refer to the Switch Task documentation.

If

If runs one branch of tasks when a condition is true and an optional else branch when it is false. The condition must evaluate to a boolean: 0, -0, null, and '' evaluate to false; all other values evaluate to true.

id: if_condition
namespace: company.team

inputs:
  - id: param
    type: BOOL

tasks:
  - id: if
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.param }}"
    then:
      - id: when_true
        type: io.kestra.plugin.core.log.Log
        message: "This is true"
    else:
      - id: when_false
        type: io.kestra.plugin.core.log.Log
        message: "This is false"

For more details, check out the If Task documentation.

Loop

The Loop task iterates over a set of values and runs child tasks for each item in an isolated sub-execution with its own context.

values accepts a list, a JSON array string, a map, or an ION file URI. When values is a URI, Kestra performs one iteration per line of the file. When values is a list or a JSON array string, Kestra drops duplicate elements, so ["a", "b", "a"] runs two iterations.

id: loop-basic
namespace: company.team

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values: ["value 1", "value 2", "value 3"]
    tasks:
      - id: log
        type: io.kestra.plugin.core.log.Log
        message: "index={{ item.index }} value={{ item.value }}"

Inside each iteration, use the item variable to access the iteration context:

For more details on item, see loop iteration context in the expressions reference.

Iterating over objects

When values contains a list of objects, each item.value is a JSON string. Use fromJson(item.value).field to read fields; item.value.field does not work.

id: parallel_tasks_example
namespace: company.team

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values:
      - { id: 101, email: "a@example.com" }
      - { id: 102, email: "b@example.com" }
    fetchType: AUTO
    outputs:
      - id: user_id
        type: INT
        value: "{{ fromJson(item.value).id }}"
      - id: email
        type: STRING
        value: "{{ fromJson(item.value).email }}"
    tasks:
      - id: log_user
        type: io.kestra.plugin.core.log.Log
        message: "User {{ fromJson(item.value).id }} -> {{ fromJson(item.value).email }}"

Concurrent execution

By default (concurrencyLimit: 0), all iterations run in parallel and workers regulate actual throughput. Set concurrencyLimit to a positive integer to cap how many iterations run at once. Any value set on the task or via a policy takes precedence over the default.

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    concurrencyLimit: 0
    tasks:
      - id: parallel
        type: io.kestra.plugin.core.flow.Parallel
        tasks:
          - id: log
            type: io.kestra.plugin.core.log.Log
            message: "Processing {{ item.value }}"
          - id: shell
            type: io.kestra.plugin.scripts.shell.Commands
            commands:
              - "echo done {{ item.value }}"

Failure propagation

By default (transmitFailed: true), a failed iteration causes the Loop task itself to fail. Set transmitFailed: false to let the loop continue even when individual iterations fail.

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values: ["ok", "fail", "ok"]
    transmitFailed: false
    tasks:
      - id: maybe_fail
        type: io.kestra.plugin.core.flow.If
        condition: '{{ item.value == "fail" }}'
        then:
          - id: do_fail
            type: io.kestra.plugin.core.execution.Fail
        else:
          - id: success
            type: io.kestra.plugin.core.log.Log
            message: "OK: {{ item.value }}"

Error handling per iteration

Use errors: to run tasks when an iteration fails, and finally: to run a block once after all iterations complete regardless of outcome. errors: runs within the failing iteration regardless of transmitFailed: with transmitFailed: true (default), the loop stops after the failing iteration completes; with transmitFailed: false, the loop continues to subsequent iterations. finally: always runs regardless.

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values:
      - ok
      - boom
      - ok
    transmitFailed: false
    tasks:
      - id: maybe_fail
        type: io.kestra.plugin.scripts.shell.Commands
        commands:
          - |
            if [ "{{ item.value }}" = "boom" ]; then
              echo "failing on {{ item.value }}" >&2
              exit 1
            fi
            echo "ok {{ item.value }}"
    errors:
      - id: handle_error
        type: io.kestra.plugin.core.log.Log
        message: "Iteration {{ item.index }} ({{ item.value }}) failed"
    finally:
      - id: cleanup
        type: io.kestra.plugin.core.log.Log
        message: "Loop completed (with or without failures)"

Stop early with breakWhen

Set breakWhen to a Pebble expression that Kestra evaluates after each iteration finishes. When the expression is truthy (true, a non-empty string, or a non-zero number), the loop starts no new iterations and counts the remaining ones as SKIPPED in terminatedIterations.

In addition to item, the expression can read the iteration that just finished:

The following example fetches pages one at a time and stops after the first page that reports no more data:

id: loop-break-when
namespace: company.team

tasks:
  - id: fetch_pages
    type: io.kestra.plugin.core.flow.Loop
    values: [1, 2, 3, 4, 5]
    concurrencyLimit: 1
    breakWhen: "{{ iteration.outputs.has_more == false }}"
    outputs:
      - id: has_more
        type: BOOL
        value: "{{ outputs.fetch.value | number < 3 }}"
    tasks:
      - id: fetch
        type: io.kestra.plugin.core.debug.Return
        format: "{{ item.value }}"

Kestra checks breakWhen only while some iterations have not started yet. With the default concurrencyLimit: 0, every iteration starts at once, so breakWhen has no effect. Set concurrencyLimit: 1 to check the condition before each new iteration starts.

To stop on a failed iteration, set transmitFailed: false and check iteration.state. With the default transmitFailed: true, the first failed iteration fails the loop before Kestra evaluates breakWhen.

Nested loops

Loops can be nested to any depth. Because item is bound to the loop execution rather than individual task runs, flowable tasks nested inside a loop can access item directly without a parent. prefix.

item.parents lists the enclosing loops from the outermost inward. item.parents[0] is the outermost loop, and the last element is the immediate parent, the same as item.parent.

tasks:
  - id: outer
    type: io.kestra.plugin.core.flow.Loop
    values: ["bucket1", "bucket2"]
    tasks:
      - id: middle
        type: io.kestra.plugin.core.flow.Loop
        values: [2025, 2026]
        tasks:
          - id: inner
            type: io.kestra.plugin.core.flow.Loop
            values: ["Jan", "Feb", "Mar"]
            tasks:
              - id: log
                type: io.kestra.plugin.core.log.Log
                message: "bucket={{ item.parents[0].value }} year={{ item.parent.value }} month={{ item.value }}"

Loop outputs

By default, task outputs produced inside a loop are not accessible to tasks that run after the loop. Use the outputs property on the Loop task to explicitly declare which values to expose.

id: loop-outputs
namespace: company.team

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values: ["a", "b", "c"]
    fetchType: AUTO
    outputs:
      - id: result
        type: STRING
        value: "{{ outputs.process.values.result }}"
    tasks:
      - id: process
        type: io.kestra.plugin.core.output.OutputValues
        values:
          result: "processed {{ item.value }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "Loop ran {{ outputs.loop.iterationCount }} iterations"

The loop also exposes monitoring outputs regardless of whether outputs is declared:

The fetchType property controls how iteration outputs are collected:

• FETCH keeps each iteration’s declared outputs inline in the execution context. Use it for small iteration counts.
• STORE writes each iteration’s declared outputs to an ION file in internal storage. Each entry in outputs.<loop_id>.outputs then holds only a uri to that file, so loopOutputs() returns null for your declared output IDs. Use it for large iteration counts.
• AUTO (the default) uses STORE when values is a URI and FETCH otherwise.

Processing large files

When values is a list of URIs from a Split task, each iteration receives one chunk URI as item.value. The following example splits a large file into chunks, processes each in parallel, and merges the results:

Passing values: "{{ outputs.split.uris }}" where outputs.split.uris is a list is different from passing a single file URI. When values is a list, each item.value is one element of that list. When values is a single URI string, Kestra iterates line-by-line through the file.

id: map-reduce
namespace: company.team

tasks:
  - id: download
    type: io.kestra.plugin.core.http.Download
    uri: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv

  - id: to_ion
    type: io.kestra.plugin.serdes.csv.CsvToIon
    from: "{{ outputs.download.uri }}"

  - id: split
    type: io.kestra.plugin.core.storage.Split
    from: "{{ outputs.to_ion.uri }}"
    rows: 25

  - id: per_chunk
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.split.uris }}"
    concurrencyLimit: 4
    fetchType: FETCH
    outputs:
      - id: data
        type: STRING
        value: "{{ outputs.aggregate.uri }}"
    tasks:
      - id: aggregate
        type: io.kestra.plugin.transform.Aggregate
        from: "{{ item.value }}"
        outputType: STORE
        groupBy: [customer_email]
        aggregates:
          orders:
            expr: count()
            type: INT
          revenue:
            expr: sum(todecimal(total))
            type: DECIMAL

  - id: concat
    type: io.kestra.plugin.core.storage.Concat
    files: "{{ loopOutputs(outputs.per_chunk.outputs, 'data') }}"
    extension: .ion

  - id: reduce
    type: io.kestra.plugin.transform.Aggregate
    from: "{{ outputs.concat.uri }}"
    outputType: STORE
    groupBy: [customer_email]
    aggregates:
      orders:
        expr: sum(orders)
        type: INT
      revenue:
        expr: sum(revenue)
        type: DECIMAL

fetchType: FETCH collects per-iteration output URIs inline; pass them to Concat via loopOutputs(outputs.per_chunk.outputs, 'data').

Accessing loop outputs in a script task

The following example runs a Python task inside a loop to compute a value, then reads the collected results in a subsequent Python task using the monitoring output and the Kestra Python SDK.

id: loop-python-outputs
namespace: company.team

tasks:
  - id: process_items
    type: io.kestra.plugin.core.flow.Loop
    values: [1, 2, 3, 4, 5]
    outputs:
      - id: squared
        type: INT
        value: "{{ outputs.compute.vars.result }}"
    tasks:
      - id: compute
        type: io.kestra.plugin.scripts.python.Script
        dependencies:
          - kestra
        script: |
          from kestra import Kestra
          n = {{ item.value }}
          Kestra.outputs({"result": n * n})

  - id: analyze
    type: io.kestra.plugin.scripts.python.Script
    dependencies:
      - kestra
    script: |
      from kestra import Kestra

      iteration_count = {{ outputs.process_items.iterationCount }}

      # outputs.process_items.outputs is a list of iteration results:
      # [{"item": {"value": "1", "iteration": 0}, "outputs": {"squared": 1}}, ...]
      all_outputs = {{ outputs.process_items.outputs | toJson }}

      squared_values = [iteration["outputs"]["squared"] for iteration in all_outputs]

      print(f"Processed {iteration_count} items")
      print(f"Squared values: {squared_values}")
      print(f"Sum of squares: {sum(squared_values)}")

      Kestra.outputs({"total": sum(squared_values)})

outputs.process_items.iterationCount is always available after the loop finishes. outputs.process_items.outputs is a list of iteration results: each entry contains an item object (value, the zero-based iteration, and key when values is a map) and an outputs map of the declared output values. To access the first iteration’s output in an expression, use outputs.process_items.outputs[0].outputs.squared. To extract one output across all iterations as a list, use the loopOutputs() function: {{ loopOutputs(outputs.process_items.outputs, 'squared') }}.

For more details, see the Loop task documentation.

LoopUntil

LoopUntil runs a group of tasks repeatedly until a boolean condition evaluates to true. After each iteration, the task evaluates the condition expression; if it evaluates to false, the block is executed again after the configured interval.

Key properties:

• condition: expression evaluated after each iteration; has access to the child task outputs from the most recent run (e.g. {{ outputs.checkStatus.code }}).
• tasks: the list of child tasks to run before re-evaluating the condition.
• checkFrequency: optional guardrails that define interval, maxIterations, and/or maxDuration between repeats. (See the LoopUntil migration note for default values.)
• failOnMaxReached: if true, the task fails when maxIterations or maxDuration is reached without the condition becoming true. Default: false (the task succeeds when limits are reached).

After the loop completes, outputs.<task_id>.iterationCount holds the total number of iterations (1-based).

Example: poll an API until it returns HTTP 200, checking every 30 seconds and stopping after 50 attempts if it never succeeds.

id: loop_until
namespace: company.team

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.LoopUntil
    condition: "{{ outputs.ping.code == 200 }}"
    checkFrequency:
      interval: PT30S
      maxIterations: 50
    tasks:
      - id: ping
        type: io.kestra.plugin.core.http.Request
        method: GET
        uri: https://kestra.io/api/mock

For more details, refer to the LoopUntil Task documentation.

AllowFailure

AllowFailure lets child tasks fail without failing the overall execution. If any child task fails:

• The AllowFailure task is marked with status WARNING.
• All child tasks inside AllowFailure stop immediately.
• The execution continues for all other tasks.
• At the end, the execution as a whole is marked as status WARNING.

In the following example:

• allow_failure will be labelled as WARNING.
• ko will be labelled as FAILED.
• next will not be run.
• end will be run and labelled SUCCESS.

id: each
namespace: company.team

tasks:
  - id: allow_failure
    type: io.kestra.plugin.core.flow.AllowFailure
    tasks:
      - id: ko
        type: io.kestra.plugin.core.execution.Fail
      - id: next
        type: io.kestra.plugin.core.log.Log
        message: "{{ task.id }} > {{ taskrun.startDate }}"

  - id: end
    type: io.kestra.plugin.core.log.Log
    message: "{{ task.id }} > {{ taskrun.startDate }}"

For more details, refer to the AllowFailure Task documentation.

Fail

Fail explicitly fails the execution, optionally guarded by a condition expression. Without a condition, Fail always fails unconditionally.

id: fail_on_switch
namespace: company.team

inputs:
  - id: param
    type: STRING
    required: true

tasks:
  - id: switch
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.param }}"
    cases:
      case1:
        - id: case1
          type: io.kestra.plugin.core.log.Log
          message: Case 1
      case2:
        - id: case2
          type: io.kestra.plugin.core.log.Log
          message: Case 2
      notexist:
        - id: fail
          type: io.kestra.plugin.core.execution.Fail
      default:
        - id: default
          type: io.kestra.plugin.core.log.Log
          message: default

With a condition, it can validate inputs before any work begins.

id: fail_on_condition
namespace: company.team

inputs:
  - id: param
    type: STRING
    required: true

tasks:
  - id: before
    type: io.kestra.plugin.core.log.Log
    message: "I'm before the fail on condition"
  - id: fail
    type: io.kestra.plugin.core.execution.Fail
    condition: "{{ inputs.param == 'fail' }}"
  - id: after
    type: io.kestra.plugin.core.log.Log
    message: "I'm after the fail on condition"

For more information, refer to the Fail Task documentation.

Subflow

Subflow starts a child execution of another flow, letting you decompose complex workflows, share reusable logic across namespaces, and monitor each execution independently.

Required properties are namespace and flowId. Pass values to the subflow via inputs; those inputs must be declared in the subflow’s definition.

By default (wait: true), the parent execution waits for the subflow to finish before continuing. Set wait: false to fire-and-forget; the parent moves on immediately without tracking the child’s result.

When wait: true, the parent captures the subflow’s final state and outputs:

• {{ outputs.subflow_task.executionId }}: the child execution ID
• {{ outputs.subflow_task.state }}: the child’s final state (SUCCESS, FAILED, etc.)
• {{ outputs.subflow_task.outputs.some_key }}: a value from the subflow’s declared outputs

transmitFailed: true (the default when wait: true) causes the parent to fail if the subflow fails. Set it to false to continue the parent regardless of the child’s outcome.

id: parent_flow
namespace: company.team

tasks:
  - id: call_subflow
    type: io.kestra.plugin.core.flow.Subflow
    namespace: company.team
    flowId: my_subflow
    inputs:
      user: "{{ inputs.username }}"
    wait: true
    transmitFailed: true

  - id: use_output
    type: io.kestra.plugin.core.log.Log
    message: "Subflow returned: {{ outputs.call_subflow.outputs.result }}"

revision pins the subflow to a specific version. inheritLabels: true forwards the parent’s execution labels to the child. scheduleDate defers the child execution to a future time.

For more details, refer to the Subflow Task documentation.

WorkingDirectory

WorkingDirectory runs all nested tasks sequentially in the same directory on the same worker, so downstream tasks can read files written by earlier ones.

id: working_dir_flow
namespace: company.team

tasks:
  - id: working_dir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
      - id: first
        type: io.kestra.plugin.scripts.shell.Commands
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        commands:
          - 'echo "{{ taskrun.id }}" > {{ workingDir }}/stay.txt'

      - id: second
        type: io.kestra.plugin.scripts.shell.Commands
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        commands:
          - |
            echo '::{"outputs": {"stay":"'$(cat {{ workingDir }}/stay.txt)'"}}::'

This task can also cache files inside the working directory, for example, to cache script dependencies like the node_modules of a node Script task.

id: node_with_cache
namespace: company.team

tasks:
  - id: working_dir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    cache:
      patterns:
        - node_modules/**
      ttl: PT1H
    tasks:
      - id: script
        type: io.kestra.plugin.scripts.node.Script
        beforeCommands:
          - npm install colors
        script: |
          const colors = require("colors");
          console.log(colors.red("Hello"));

This task can also fetch files from namespace files and make them available to all child tasks.

id: node_with_cache
namespace: company.team

tasks:
  - id: working_dir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    namespaceFiles:
      enabled: true
      include:
        - dir1/*.*
      exclude:
        - dir2/*.*
    tasks:
      - id: shell
        type: io.kestra.plugin.scripts.shell.Commands
        commands:
          - cat dir1/file1.txt

For more details, refer to the WorkingDirectory Task documentation.

Pause

Pause halts the execution until it is manually resumed or a timeout expires. Tasks declared after the Pause in the flow run once the execution resumes.

A paused execution can be resumed manually from the Gantt tab on the execution page, or via the API at POST /api/v1/executions/{executionId}/resume.

id: pause_for_approval
namespace: company.team

tasks:
  - id: before
    type: io.kestra.plugin.core.log.Log
    message: "Waiting for manual approval"

  - id: approval
    type: io.kestra.plugin.core.flow.Pause

  - id: after
    type: io.kestra.plugin.core.log.Log
    message: "Approved — continuing execution"

pauseDuration resumes the execution automatically after a fixed interval (ISO 8601 duration format). The behavior property controls what happens when that duration expires without a manual resume: RESUME continues (the default), WARN continues with a warning, CANCEL cancels the execution, or FAIL fails the task.

tasks:
  - id: wait
    type: io.kestra.plugin.core.flow.Pause
    pauseDuration: PT5M
    behavior: WARN

onResume accepts structured input from the person approving the pause. Downstream tasks access those values via {{ outputs.<pause_task_id>.onResume.<input_id> }}.

tasks:
  - id: wait_for_approval
    type: io.kestra.plugin.core.flow.Pause
    onResume:
      - id: approved
        description: Approve or reject
        type: BOOL
        defaults: true
      - id: reason
        description: Reason for decision
        type: STRING

  - id: log_decision
    type: io.kestra.plugin.core.log.Log
    message: "Decision: {{ outputs.wait_for_approval.onResume.approved }} — {{ outputs.wait_for_approval.onResume.reason }}"

For more details, refer to the Pause Task documentation.

DAG

DAG lets you declare tasks and their dependsOn links; Kestra derives execution order and runs tasks in parallel as their dependencies are satisfied.

Tasks with no dependsOn start immediately. The concurrent property caps how many tasks run at once (0 = no limit, the default).

Note: UI no-code forms are not available for DAG tasks; configure them in YAML or the code editor.

id: dag_flow
namespace: company.team

tasks:
  - id: dag
    type: io.kestra.plugin.core.flow.Dag
    tasks:
      - task:
          id: task1
          type: io.kestra.plugin.core.log.Log
          message: task 1
      - task:
          id: task2
          type: io.kestra.plugin.core.log.Log
          message: task 2 (depends on task1)
        dependsOn:
          - task1
      - task:
          id: task3
          type: io.kestra.plugin.core.log.Log
          message: task 3 (depends on task1)
        dependsOn:
          - task1
      - task:
          id: task4
          type: io.kestra.plugin.core.log.Log
          message: task 4 (depends on task2 and task3)
        dependsOn:
          - task2
          - task3

In this example, task2 and task3 run in parallel after task1 completes, and task4 starts once both finish.

For more details, refer to the DAG Task documentation.