docstutorial

Triggers​Triggers

Triggers automatically start your flow based on events.

A trigger can be a scheduled date, a new file arrival, a new message in a queue, or the end of another flow's execution.

Defining triggers

Use the triggers keyword in the flow and define a list of triggers. You can have several triggers attached to a flow.

The trigger definition looks similar to the task definition — it contains an id, a type, and additional properties related to the specific trigger type.

The workflow below is automatically triggered every day at 10 AM, as well as anytime when the first_flow finishes its execution. Both triggers are independent of each other.

yaml
id: getting_started
namespace: company.team

tasks:
  - id: hello_world
    type: io.kestra.plugin.core.log.Log
    message: Hello World!

triggers:
  - id: schedule_trigger
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 10 * * *

  - id: flow_trigger
    type: io.kestra.plugin.core.trigger.Flow
    conditions:
      - type: io.kestra.plugin.core.condition.ExecutionFlow
        namespace: company.team
        flowId: first_flow

Add a trigger to your flow

Building on our example flow from the previous pages, let's add one of the above triggers to the flow.

yaml
triggers:
  - id: every_monday_at_10_am
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 10 * * 1

Our getting_started flow now runs every Monday at 10AM, starting the week with the new product data.

yaml
id: getting_started
namespace: company.team

labels:
  owner: engineering

tasks:
  - id: api
    type: io.kestra.plugin.core.http.Request
    uri: https://dummyjson.com/products

  - id: python
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:slim
    beforeCommands:
      - pip install polars
    warningOnStdErr: false
    outputFiles:
      - "products.csv"
    script: |
      import polars as pl
      data = {{ outputs.api.body | jq('.products') | first }}
      df = pl.from_dicts(data)
      df.glimpse()
      df.select(["brand", "price"]).write_csv("products.csv")

  - id: sqlQuery
    type: io.kestra.plugin.jdbc.duckdb.Query
    inputFiles:
      in.csv: "{{ outputs.python.outputFiles['products.csv'] }}"
    sql: |
      SELECT brand, round(avg(price), 2) as avg_price
      FROM read_csv_auto('{{ workingDir }}/in.csv', header=True)
      GROUP BY brand
      ORDER BY avg_price DESC;
    store: true

triggers:
  - id: every_monday_at_10_am
    type: io.kestra.plugin.core.trigger.Schedule
    cron: 0 10 * * 1

To learn more about triggers, check out the full triggers documentation.

Next, let's orchestrate more complex workflows

Was this page helpful?