Best Practices for the Loop Task

Use Loop for all iteration needs in Kestra.

Choose the right iteration pattern

Loop runs child tasks for each item in a list, map, file, or URI list. Every iteration is an isolated sub-execution.

Use a plain Loop when:

• You need to run the same tasks for each item in a list or dataset.
• Each iteration should process one value, one object, or one file chunk.
• You want parallel iteration with controlled concurrency.
• You want per-iteration failure handling without stopping the entire loop.

Use Loop + Subflow when:

• You need full execution isolation per batch — own retries, own logs, own failure state.
• Each batch should be independently restartable.

Access the iteration value

Inside a Loop, use item.value for the current value and item.index for the zero-based position. These are available in every child task, including those nested inside If, Parallel, or other flowable tasks — no parent traversal needed.

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    values: ["north", "south", "west"]
    tasks:
      - id: log
        type: io.kestra.plugin.core.log.Log
        message: "region={{ item.value }} index={{ item.index }}"

When iterating over JSON objects, item.value is a JSON string. Use fromJson(item.value).field to access properties — item.value.field does not work.

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

Expose outputs explicitly

Task outputs inside a loop are not visible outside it by default. Declare an outputs: block on the Loop task to surface values. Choose fetchType based on data volume:

• AUTO: default, switches automatically based on whether values is a URI
• FETCH: collects all iteration results inline (suitable for small iteration counts)
• STORE: writes each iteration’s outputs to an ION file in internal storage and keeps only its uri (preferred for large iteration counts). loopOutputs() cannot read declared output IDs in this mode.

After the loop:

• outputs.<loop_id>.outputs is a list of per-iteration results, in the order the iterations finished
• item.iteration on each entry identifies the iteration; filter on it to read a specific iteration, because outputs[n] is the nth iteration to finish
• loopOutputs(outputs.<loop_id>.outputs, '<output_id>') extracts one field across all iterations as a flat list, also in completion order

Example: collect outputs and read them downstream

id: loop_outputs
namespace: company.team

tasks:
  - id: enrich_regions
    type: io.kestra.plugin.core.flow.Loop
    values: ["north", "south", "west"]
    concurrencyLimit: 2
    fetchType: AUTO
    outputs:
      - id: bucket
        type: STRING
        value: "landing-{{ item.value }}"
      - id: message
        type: STRING
        value: "{{ outputs.build_message.values.result }}"
    tasks:
      - id: build_message
        type: io.kestra.plugin.core.output.OutputValues
        values:
          result: "Load {{ item.value }} into landing-{{ item.value }}"

  - id: log_first
    type: io.kestra.plugin.core.log.Log
    message: "{{ outputs.enrich_regions.outputs | jq('.[] | select(.item.iteration == 0) | .outputs.message') | first }}"

  - id: log_all
    type: io.kestra.plugin.core.log.Log
    message: "{{ loopOutputs(outputs.enrich_regions.outputs, 'message') }}"

Inside the loop, sibling task outputs are accessed with plain outputs.task_id.attribute syntax — each iteration runs in its own isolated context, so there is no ambiguity.

Use concurrencyLimit deliberately

• 0 (default): unlimited; all iterations run simultaneously and workers regulate actual throughput
• A positive integer: bounded parallelism; set this for heavy workloads to avoid overwhelming workers
• 1: sequential execution; use when order matters or resources are constrained

Any value set on the task or via a policy takes precedence over the default.

Process large files with Split and Loop

For file-backed datasets, use Split to break the file into chunk URIs, then loop over the URI list. Each item.value is one chunk URI.

Passing values: "{{ outputs.split.uris }}" where outputs.split.uris is a list is different from passing a single file URI string. A list iterates over elements; a single URI string iterates line-by-line through that file.

tasks:
  - id: split
    type: io.kestra.plugin.core.storage.Split
    from: "{{ inputs.file }}"
    rows: 100

  - id: per_chunk
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.split.uris }}"
    concurrencyLimit: 4
    fetchType: FETCH
    outputs:
      - id: result_uri
        type: STRING
        value: "{{ outputs.process.values.result }}"
    tasks:
      - id: process
        type: io.kestra.plugin.core.output.OutputValues
        values:
          result: "processed chunk {{ item.index }}: {{ item.value }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "{{ loopOutputs(outputs.per_chunk.outputs, 'result_uri') }}"

Map-reduce: collect and aggregate file outputs across iterations

Use the map-reduce pattern when each iteration produces a file output that you need to combine and reduce after the loop.

The structure is always the same:

1. Split — break the input into chunk URIs
2. Loop — process each chunk in parallel; declare a file URI as a STRING output
3. Concat — stitch the per-iteration output files into one
4. Aggregate (or any reduce task) — merge results across chunks

The key is declaring the per-iteration output URI as a STRING in the Loop outputs: block, then using loopOutputs() to extract that list and pass it directly to Concat.

id: elt_csv_split_loop_aggregate
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 }}"
    header: true

  - 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: filter
        type: io.kestra.plugin.transform.Filter
        from: "{{ item.value }}"
        where: todecimal(total) > 10

      - id: aggregate
        type: io.kestra.plugin.transform.Aggregate
        from: "{{ outputs.filter.uri }}"
        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

Key points:

• The Loop output declares type STRING even though the value is a file URI. loopOutputs() extracts strings, and Concat accepts a list of URI strings.
• fetchType: FETCH keeps the per-iteration output objects in memory so loopOutputs() can read them immediately after the loop. Keep FETCH for this pattern: with STORE, each iteration’s outputs are replaced by a file uri, and loopOutputs() returns null for data.
• The final reduce task re-aggregates because the same key (e.g. a customer) may appear in multiple chunks. The two-pass structure — per-chunk aggregate then cross-chunk reduce — is the canonical map-reduce pattern.

Handle per-iteration failures

Set transmitFailed: false to continue the loop when individual iterations fail. Use errors: to run tasks per failed iteration, and finally: for a one-time cleanup block after all iterations finish. errors: runs on failure regardless of transmitFailed — with transmitFailed: false, the loop continues to subsequent iterations after handling the error. 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 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 finished (with or without failures)"

Stop the loop early with breakWhen

Set breakWhen to stop starting new iterations once a condition holds. Kestra evaluates the expression after each iteration finishes and counts the remaining iterations as SKIPPED. The following example stops at the first failed iteration:

tasks:
  - id: loop
    type: io.kestra.plugin.core.flow.Loop
    concurrencyLimit: 1
    values: ["ok", "fail", "ok", "ok"]
    transmitFailed: false
    breakWhen: "{{ iteration.state == 'FAILED' }}"
    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

Two settings make this work:

• concurrencyLimit: 1: Kestra checks breakWhen only while some iterations have not started yet. With the default 0, every iteration starts at once and breakWhen has no effect.
• transmitFailed: false: with the default true, the first failed iteration fails the loop before Kestra evaluates breakWhen.

For the full list of variables available in breakWhen, see Stop early with breakWhen.

Use Loop + Subflow for isolated per-batch execution

When each batch needs its own execution — independent retries, logs, and failure ownership — pair Loop with Subflow. The parent splits the data and fans out; the child flow receives one chunk URI per invocation and returns its result as a flow-level output.

# Parent flow
tasks:
  - id: split
    type: io.kestra.plugin.core.storage.Split
    from: "{{ inputs.file }}"
    rows: 100

  - id: per_batch
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ outputs.split.uris }}"
    concurrencyLimit: 4
    fetchType: FETCH
    outputs:
      - id: result_uri
        type: STRING
        value: "{{ outputs.run_child.outputs.uri }}"
    tasks:
      - id: run_child
        type: io.kestra.plugin.core.flow.Subflow
        namespace: company.team
        flowId: process_batch
        wait: true
        transmitFailed: true
        inputs:
          batch_uri: "{{ item.value }}"

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

Compose Loop with supporting tasks

Loop iterates. These tasks handle the rest — batching, transforming, stitching, and reducing. Each does one thing well; combine them around Loop to build larger pipelines.

Common mistakes to avoid

• Do not use taskrun.value or taskrun.iteration — use item.value and item.index.
• Do not access item.value.field directly on object values — use fromJson(item.value).field.
• Do not expect loop outputs to be visible downstream without declaring an outputs: block.
• Do not expect duplicate list values to run twice. Kestra drops duplicate elements from a list or JSON array.
• Do not use breakWhen with the default concurrencyLimit: 0. Every iteration starts at once, so the condition never stops anything.
• Do not use outputs.task_id[item.value] inside a loop — sibling outputs are accessed with plain outputs.task_id.attribute.
• For very large datasets, set an explicit concurrencyLimit to cap parallelism. The default unlimited behavior can overwhelm workers if iterations are resource-intensive.

For more details, see the Loop task documentation and the Flowable Tasks reference.