Core Plugins and tasks ForEachItem

Core Plugins and tasks ForEachItem
Certified
Spawn a subflow for each batch of items.

Reads items from a Kestra internal storage URI (newline JSON, Ion, CSV, etc.), optionally splits into batches, and starts one subflow execution per batch.

Special variables: taskrun.items (URI of the batch file) and taskrun.iteration (batch index). If the parent flow restarts, already-started subflows are restarted too.

yaml
type: io.kestra.plugin.core.flow.ForEachItem

Examples

Execute a subflow for each batch of items. The subflow orders is called from the parent flow orders_parallel using the ForEachItem task in order to start one subflow execution for each batch of items.

id: orders
namespace: company.team

inputs: 
  - id: order
    type: STRING

tasks: 
  - id: read_file
    type: io.kestra.plugin.scripts.shell.Commands
    taskRunner: 
      type: io.kestra.plugin.core.runner.Process
    commands: 
      - cat "{{ inputs.order }}"

  - id: read_file_content
    type: io.kestra.plugin.core.log.Log
    message: "{{ read(inputs.order) }}"

yaml
id: orders_parallel
namespace: company.team

tasks:
  - id: extract
    type: io.kestra.plugin.jdbc.duckdb.Query
    sql: |
      INSTALL httpfs;
      LOAD httpfs;
      SELECT *
      FROM read_csv_auto('https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv', header=True);
    store: true

  - id: each
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{ outputs.extract.uri }}"
    batch:
      rows: 1
    namespace: company.team
    flowId: orders
    wait: true # wait for the subflow execution
    transmitFailed: true # fail the task run if the subflow execution fails
    inputs:
      order: "{{ taskrun.items }}" # special variable that contains the items of the batch

Execute a subflow for each JSON item fetched from a REST API. The subflow mysubflow is called from the parent flow iterate_over_json using the ForEachItem task; this creates one subflow execution for each JSON object.

Note how we first need to convert the JSON array to JSON-L format using the JsonWriter task. This is because the items attribute of the ForEachItem task expects a file where each line represents a single item. Suitable file types include Amazon ION (commonly produced by Query tasks), newline-separated JSON files, or CSV files formatted with one row per line and without a header. For other formats, you can use the conversion tasks available in the io.kestra.plugin.serdes module.

In this example, the subflow mysubflow expects a JSON object as input. The JsonReader task first reads the JSON array from the REST API and converts it to ION. Then, the JsonWriter task converts that ION file to JSON-L format, suitable for the ForEachItem task.

id: mysubflow
namespace: company.team

inputs: 
  - id: json
    type: JSON

tasks: 
  - id: debug
    type: io.kestra.plugin.core.log.Log
    message: "{{ inputs.json }}"

yaml
id: iterate_over_json
namespace: company.team

tasks:
  - id: download
    type: io.kestra.plugin.core.http.Download
    uri: "https://api.restful-api.dev/objects"
    contentType: application/json
    method: GET
    failOnEmptyResponse: true
    timeout: PT15S

  - id: json_to_ion
    type: io.kestra.plugin.serdes.json.JsonToIon
    from: "{{ outputs.download.uri }}"
    newLine: false # regular json

  - id: ion_to_jsonl
    type: io.kestra.plugin.serdes.json.IonToJson
    from: "{{ outputs.json_to_ion.uri }}"
    newLine: true # JSON-L

  - id: for_each_item
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{ outputs.ion_to_jsonl.uri }}"
    batch:
      rows: 1
    namespace: company.team
    flowId: mysubflow
    wait: true
    transmitFailed: true
    inputs:
      json: "{{ json(read(taskrun.items)) }}"

This example shows how to use the combination of ForEach and ForEachItem tasks to process files from an S3 bucket. The ForEach iterates over files from the S3 trigger, and the ForEachItem task is used to split each file into batches. The process_batch subflow is then called with the data input parameter set to the URI of the batch to process.

id: process_batch
namespace: company.team

inputs: 
  - id: data
    type: FILE

tasks: 
  - id: debug
    type: io.kestra.plugin.core.log.Log
    message: "{{ read(inputs.data) }}"

yaml
id: process_files
namespace: company.team

tasks:
  - id: loop_over_files
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ trigger.objects | jq('.[].uri') }}"
    tasks:
      - id: subflow_per_batch
        type: io.kestra.plugin.core.flow.ForEachItem
        items: "{{ trigger.uris[parent.taskrun.value] }}"
        batch:
          rows: 1
        flowId: process_batch
        namespace: company.team
        wait: true
        transmitFailed: true
        inputs:
          data: "{{ taskrun.items }}"

triggers:
  - id: s3
    type: io.kestra.plugin.aws.s3.Trigger
    interval: "PT1S"
    accessKeyId: "<access-key>"
    secretKeyId: "<secret-key>"
    region: "us-east-1"
    bucket: "my_bucket"
    prefix: "sub-dir"
    action: NONE

Properties

flowId *string

Min length1

The identifier of the subflow to be executed

id *string

Validation RegExp^[a-zA-Z0-9][a-zA-Z0-9_-]*

Min length1

Max length256

items *string

Min length1

The items to be split into batches and processed – make sure to set it to Kestra's internal storage URI. This can be either the output from a previous task, formatted as {{ outputs.task_id.uri }}, or a FILE type input parameter, like {{ inputs.myfile }}. This task is optimized for files where each line represents a single item. Suitable file types include Amazon ION-type files (commonly produced by Query tasks), newline-separated JSON files, or CSV files formatted with one row per line and without a header. For files in other formats such as Excel, CSV, Avro, Parquet, XML, or JSON, it's recommended to first convert them to the ION format. This can be done using the conversion tasks available in the io.kestra.plugin.serdes module, which will transform files from their original format to ION.

namespace *string

Min length1

The namespace of the subflow to be executed

allowFailure boolean

Defaultfalse

allowWarning boolean

Defaultfalse

assets

batch

Default

{
  "rows": "1",
  "separator": "\n"
}

How to split the items into batches

description string

disabled boolean

Defaultfalse

errors array

List of tasks to run if any tasks failed on this FlowableTask.

finally array

inheritLabels boolean

Defaultfalse

Flag specifying whether the subflow should inherit labels from the parent execution that triggered it.

By default, labels are not passed to the subflow execution. If you set this option to true, the child flow execution will inherit all labels from the parent execution.

inputs object

The inputs to pass to the subflow to be executed

labels arrayobject

The labels to pass to the subflow to be executed

logLevel string

Possible Values

ERRORWARNINFODEBUGTRACE

logToFile boolean

Defaultfalse

pluginDefaultsRef string

Reference (ref) of the pluginDefaults to apply to this task.

restartBehavior string

DefaultRETRY_FAILED

Possible Values

NEW_EXECUTIONRETRY_FAILED

What action to take when a failed execution is restarting

RETRY_FAILED (default): will restart the each subflow executions that are failed.
NEW_EXECUTION: will create a new subflow execution for each batch of items.""

retry

Retry

Retry policy applied when the task fails.

revision integer

The revision of the subflow to be executed

By default, the last, i.e. the most recent, revision of the subflow is executed.

runIf string

Defaulttrue

scheduleDate string

Don't trigger the subflow now but schedule it on a specific date.

taskCache

timeout string

transmitFailed boolean

Defaulttrue

Flag specifying whether to fail the current execution if the subflow execution fails or is killed.

Note that this option only works if wait is set to true.

version string

Plugin Version

Defines the version of the plugin to use.

The version must follow the Semantic Versioning (SemVer) specification:

A single-digit MAJOR version (e.g., 1).
A MAJOR.MINOR version (e.g., 1.1).
A MAJOR.MINOR.PATCH version, optionally with any qualifier (e.g., 1.1.2, 1.1.0-SNAPSHOT).