Outputs

Outputs let you pass data between tasks and flows.

What are outputs

A workflow execution can generate outputs. Outputs are stored in the flow’s execution context and can be accessed by all downstream tasks and flows.

Each task defines its own output attributes — see the task’s documentation for details.

You can retrieve outputs from other tasks within all dynamic properties.

Using outputs

Below is an example of how to use the output of the produce_output task in the use_output task. We use the Return task that has one output attribute named value.

id: task_outputs_example
namespace: company.team

tasks:
  - id: produce_output
    type: io.kestra.plugin.core.debug.Return
    format: my output {{ execution.id }}

  - id: use_output
    type: io.kestra.plugin.core.log.Log
    message: The previous task output is {{ outputs.produce_output.value }}

In this example, the first task produces an output from the format property. This output attribute is then used in the second task message property.

The expression {{ outputs.produce_output.value }} references the previous task output attribute.

The Outputs tab shows the output for produce_output task. There is no output for use_output task as it only logs a message.

In the next example, we can see a file is passed between an input and a task, where the task generates a new file as an output:

id: bash_with_files
namespace: company.team

description: This flow shows how to pass files between inputs and tasks in Shell scripts.

inputs:
  - id: file
    type: FILE

tasks:
  - id: rename
    type: io.kestra.plugin.scripts.shell.Commands
    commands:
      - mv file.tmp output.tmp
    inputFiles:
      file.tmp: "{{ inputs.file }}"
    outputFiles:
      - "*.tmp"

Internal storage

Each task can store data in Kestra’s internal storage. If an output is stored in internal storage, it contains a URI pointing to the file location. This output attribute could be used by other tasks to access the stored data.

The following example stores the query results in internal storage, then accesses it in the write_to_csv task:

id: output_sample
namespace: company.team

tasks:
  - id: output_from_query
    type: io.kestra.plugin.gcp.bigquery.Query
    sql: |
      SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2023`
      WHERE DATE(datehour) = current_date()
      ORDER BY datehour desc, views desc
      LIMIT 10
    store: true

  - id: write_to_csv
    type: io.kestra.plugin.serdes.csv.IonToCsv
    from: "{{ outputs.output_from_query.uri }}"

Flow outputs

Since Kestra 0.15.0, the flow can also produce strongly typed outputs. You can add them using the outputs attribute in the flow definition.

Here is an example of a flow that produces an output:

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 }}"

Outputs are defined as a list of key-value pairs. The id is the name of the output attribute (must be unique within a flow), and the value is the value of the output. You can also add a description to the output.

Flow outputs appear in the Overview tab of the Executions page.

Pass data between flows using flow outputs

Here is how you can access the flow output in a parent flow:

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 }}"

In the example above, the subflow task produces an output attribute final. This output attribute is then used in the log_subflow_output task.

Here is what you will see in the Outputs tab of the Executions page in the parent flow:

Return outputs conditionally

You can return different outputs based on conditions. For instance, if a given task is skipped, you may want to return a fallback value or return the output of another task. Here is an example of how you can achieve this:

id: conditionally_return_output
namespace: company.team

inputs:
  - id: run_task
    type: BOOLEAN
    defaults: true

tasks:
  - id: main
    type: io.kestra.plugin.core.debug.Return
    format: Hello World!
    runIf: "{{ inputs.run_task }}"

  - id: fallback
    type: io.kestra.plugin.core.debug.Return
    format: fallback output

outputs:
  - id: flow_output
    type: STRING
    value: "{{ tasks.main.state != 'SKIPPED' ? outputs.main.value : outputs.fallback.value }}"

Note how the Ternary Operator {{ condition ? value_if_true : value_if_false }} is used in the output expression {{ tasks.main.state != 'SKIPPED' ? outputs.main.value : outputs.fallback.value }} to return the output of the main task if it is not skipped, otherwise, it returns the output of the fallback task.

Dynamic variables (Each tasks)

Current taskrun value

In dynamic flows (for example, with an Each loop), variables are passed to tasks dynamically. You can access the current taskrun value with {{ taskrun.value }} like this:

id: taskrun_value_example
namespace: company.team

tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["value 1", "value 2", "value 3"]
    tasks:
      - id: inner
        type: io.kestra.plugin.core.debug.Return
        format: "{{ task.id }} > {{ taskrun.value }} > {{ taskrun.startDate }}"

The Outputs tab contains the output for each of the inner task.

Loop over a list of JSON objects

Within the loop, the value is always a JSON string, so the {{ taskrun.value }} is the current element as JSON string. To access properties, you need to wrap it in the fromJson() function to have a JSON object allowing to access each property easily.

id: loop_sequentially_over_list
namespace: company.team

tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values:
      - {"key": "my-key", "value": "my-value"}
      - {"key": "my-complex", "value": {"sub": 1, "bool": true}}
    tasks:
      - id: inner
        type: io.kestra.plugin.core.debug.Return
        format: "{{ fromJson(taskrun.value).key }} > {{ fromJson(taskrun.value).value }}"

Specific outputs for dynamic tasks

Dynamic tasks are tasks that run other tasks a certain number of times. A dynamic task runs multiple iterations of a set of sub-tasks.

For example, ForEach produces other tasks dynamically depending on its values property.

It is possible to reach each iteration output of dynamic tasks by using the following syntax:

id: output_sample
namespace: company.team

tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["s1", "s2", "s3"]
    tasks:
      - id: sub
        type: io.kestra.plugin.core.debug.Return
        format: "{{ task.id }} > {{ taskrun.value }} > {{ taskrun.startDate }}"

  - id: use
    type: io.kestra.plugin.core.debug.Return
    format: "Previous task produced output: {{ outputs.sub.s1.value }}"

The outputs.sub.s1.value variable reaches the value of the sub task of the s1 iteration.

Previous task lookup

It is also possible to locate a specific dynamic task by its value:

id: dynamic_looping
namespace: company.team

tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["value 1", "value 2", "value 3"]
    tasks:
      - id: inner
        type: io.kestra.plugin.core.debug.Return
        format: "{{ taskrun.value }}"

  - id: end
    type: io.kestra.plugin.core.debug.Return
    format: "{{ task.id }} > {{ outputs.inner['value 1'].value }}"

It uses the format outputs.TASKID[VALUE].ATTRIBUTE. The special bracket [] in [VALUE] is called the subscript notation; it enables using special chars like space or ’-’ in task identifiers or output attributes.

Lookup in sibling tasks

Sometimes it is useful to access outputs from other tasks in the same task tree, known as sibling tasks.

If the task tree is static, for example when using the Sequential task, you can use the {{ outputs.sibling.value }} notation where siblingis the identifier of the sibling task.

If the task tree is dynamic, for example when using the ForEach task, you need to use {{ sibling[taskrun.value] }} to access the current tree task. taskrun.value is a special variable that holds the current value of the ForEach task.

For example:

id: loop_with_sibling_tasks
namespace: company.team

tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["value 1", "value 2", "value 3"]
    tasks:
      - id: first
        type: io.kestra.plugin.core.debug.Return
        format: "{{ taskrun.value }}"

      - id: second
        type: io.kestra.plugin.core.debug.Return
        format: "{{ outputs.first[taskrun.value].value }}"

  - id: end
    type: io.kestra.plugin.core.debug.Return
    format: "{{ task.id }} > {{ outputs.second['value 1'].value }}"

When there are multiple levels of EachSequential tasks, you can use the parents variable to access the taskrun.value of the parent of the current EachSequential. For example, for two levels of EachSequential you can use outputs.sibling[parents[0].taskrun.value][taskrun.value].value.

The latter can become very complex when parents exist (multiple imbricated EachSequential). For this, you can use the currentEachOutput() function. No matter the number of parents, the following example will retrieve the correct output attribute: currentEachOutput(outputs.sibling).value thanks to this function.

Outputs Preview

Kestra provides a preview option for output files stored in internal storage. The following flow demonstrates this feature:

id: get_employees
namespace: company.team

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

On flow execution, the file is downloaded into the Kestra internal storage. When you go to the Outputs tab for this execution, the uri attribute of the download task contains the file location on Kestra’s internal storage and has a Download and a Preview button.

On clicking the Preview button, you can preview the contents of the file in a tabular format, making it extremely easy to check the contents of the file without downloading it.

Using Debug Expression

You can evaluate the output further using the Debug Expression functionality in the Outputs tab. Consider the following flow:

id: json_values
namespace: company.team

tasks:
- id: sample_json
  type: io.kestra.plugin.core.debug.Return
  format: '{"data": [1, 2, 3]}'

When you run this flow, the Outputs tab will contain the output for the sample_json task, as shown below:

You can select the task from the drop-down menu. Here, we select “sample_json” and select Debug Expression:

You can now use Pebble expressions to evaluate and analyze the output data further.

Encrypted Outputs from Script Tasks

For script task Outputs that have sensitive values, you can protect the information by using the encryptedOutputs syntax such as ::{"encryptedOutputs":{"encrypted":"my secret value"}}::.

In the following flow, the encrypted output is not shown in plain text in the Outputs UI.

id: encryped_output
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.scripts.shell.Script
    script: |
      echo '::{"outputs":{"plaintext":"plaintext_value"}}::'
      echo '::{"encryptedOutputs":{"encrypted":"my secret value"}}::'

  - id: print
    type: io.kestra.plugin.core.log.Log
    message: "{{ outputs.hello['vars']['encrypted'] }}"

The encrypted output is displayed encoded: