id: my-first-workflow
namespace: company.team
description: "Basic ETL pipeline example"
 
# Triggers — when to run this flow
triggers:
  - id: daily-schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * *"    # Every day at 9am
 
# Tasks — what to do
tasks:
  - id: extract-data
    type: io.kestra.plugin.scripts.python.Script
    script: |
      import requests
      data = requests.get("https://api.example.com/data").json()
      print(data)
 
  - id: transform-data
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # Transform logic here
      print("Transforming...")
 
  - id: load-data
    type: io.kestra.plugin.jdbc.postgresql.Query
    sql: "INSERT INTO table SELECT * FROM staging"

triggers:
  # Every day at midnight
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 0 * * *"
 
  # Every 15 minutes
  - id: frequent
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
 
  # First Monday of each month
  - id: monthly
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * 1#1"
 
  # Webhook — POST to trigger
  - id: webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "secret-key-here"

tasks:
  # Python script
  - id: python-task
    type: io.kestra.plugin.scripts.python.Script
    script: |
      import pandas as pd
      df = pd.read_csv("data.csv")
      print(df.head())
 
  # Bash / Shell
  - id: bash-task
    type: io.kestra.plugin.scripts.shell.Script
    script: |
      echo "Hello from Bash"
      curl -X POST https://api.example.com/webhook
 
  # Node.js
  - id: node-task
    type: io.kestra.plugin.scripts.node.Script
    script: |
      const axios = require('axios');
      const data = await axios.get('https://api.example.com');
      console.log(data.data);
 
  # Docker container
  - id: docker-task
    type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: "python:3.11"
    commands:
      - python -c "print('Running in Docker')"

tasks:
  # Run multiple tasks in parallel
  - id: parallel-ingest
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: ingest-source-a
        type: io.kestra.plugin.scripts.python.Script
        script: |
          print("Ingesting from source A")
 
      - id: ingest-source-b
        type: io.kestra.plugin.scripts.python.Script
        script: |
          print("Ingesting from source B")
 
      - id: ingest-source-c
        type: io.kestra.plugin.scripts.python.Script
        script: |
          print("Ingesting from source C")
 
  # All 3 run at the same time, next task waits for all to finish
  - id: merge-results
    type: io.kestra.plugin.scripts.python.Script
    script: |
      print("All sources ingested, merging...")

tasks:
  - id: risky-task
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # This might fail
      raise Exception("Simulated failure")
 
errors:
  # On failure — create a Jira ticket
  - id: alert-on-failure
    type: io.kestra.plugin.jira.issues.Create
    domain: "yourcompany.atlassian.net"
    username: "{{ secret('JIRA_USER') }}"
    apiToken: "{{ secret('JIRA_TOKEN') }}"
    projectKey: "OPS"
    summary: "Workflow {{ flow.id }} failed"
    description: "Execution {{ execution.id }} failed at task {{ task.id }}"

tasks:
  - id: check-data-quality
    type: io.kestra.plugin.scripts.python.Script
    script: |
      row_count = 1000
      print(row_count)
 
  - id: branch-on-quality
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs['check-data-quality'].vars.row_count > 500 }}"
    then:
      - id: proceed-to-load
        type: io.kestra.plugin.scripts.python.Script
        script: print("Data quality OK — proceeding")
    else:
      - id: alert-bad-data
        type: io.kestra.plugin.slack.IncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          { "text": "Data quality failed — row count too low" }

id: full-etl-pipeline
namespace: data.production
 
tasks:
  # Step 1 — Sync data from source using Airbyte
  - id: airbyte-sync
    type: io.kestra.plugin.airbyte.connections.Sync
    connectionId: "your-airbyte-connection-id"
    url: "http://airbyte:8006"
 
  # Step 2 — Run dbt transformations
  - id: dbt-run
    type: io.kestra.plugin.dbt.cli.DbtCLI
    commands:
      - dbt run --project-dir dbt_project
      - dbt test --project-dir dbt_project
 
  # Step 3 — Verify row counts in Postgres
  - id: verify-load
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('POSTGRES_URL') }}"
    sql: "SELECT COUNT(*) as row_count FROM public.transformed_data"
 
  # Step 4 — Alert Slack on completion
  - id: notify-success
    type: io.kestra.plugin.slack.IncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      { "text": "ETL pipeline completed: {{ outputs['verify-load'].rows[0]['row_count'] }} rows loaded" }

tasks:
  - id: reliable-task
    type: io.kestra.plugin.scripts.python.Script
    retry:
      type: constant
      maxAttempts: 3
      interval: PT30S    # Wait 30s between retries
    timeout: PT10M       # Fail task if it runs > 10 minutes
    script: |
      # Your resilient task code here
      print("Running with retry + timeout protection")

# WRONG — never do this
tasks:
  - id: bad-example
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "postgresql://user:PASSWORD123@host:5432/db"
 
# CORRECT — use secrets
tasks:
  - id: good-example
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('POSTGRES_URL') }}"
 
# Secrets stored in:
#  → Kestra's built-in secrets store
#  → AWS Secrets Manager
#  → Azure Key Vault
#  → HashiCorp Vault
#  → GCP Secret Manager

# docker-compose.yml — Kestra + Postgres
version: "3"
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: kestra
      POSTGRES_USER: kestra
      POSTGRES_PASSWORD: k3str4
    volumes:
      - postgres-data:/var/lib/postgresql/data
 
  kestra:
    image: kestra/kestra:latest
    pull_policy: always
    entrypoint: /bin/bash
    command: -c 'kestra server standalone'
    environment:
      KESTRA_CONFIGURATION: |
        datasources:
          postgres:
            url: jdbc:postgresql://postgres:5432/kestra
            username: kestra
            password: k3str4
    ports:
      - "8080:8080"    # Kestra UI
      - "8081:8081"    # Kestra API
    depends_on:
      - postgres
 
volumes:
  postgres-data:

# Start Kestra
docker-compose up -d
 
# Access UI at http://localhost:8080
# Access API at http://localhost:8081

# Add Kestra Helm chart
helm repo add kestra https://helm.kestra.io
helm repo update
 
# Install with custom values
helm install kestra kestra/kestra \
  --set configuration.kestra.datasources.postgres.url=jdbc:postgresql://postgres:5432/kestra \
  --set configuration.kestra.datasources.postgres.username=kestra \
  --set configuration.kestra.datasources.postgres.password=your-password
 
# Scale workers independently from the executor
kubectl scale deployment kestra-worker --replicas=5

# .github/workflows/deploy-kestra.yml
name: Deploy Kestra Flows
 
on:
  push:
    branches: [main]
 
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
 
      - name: Deploy flows to Kestra
        uses: kestra-io/deploy-action@master
        with:
          resource: flow
          directory: ./flows
          namespace: company.production
          server: ${{ secrets.KESTRA_HOSTNAME }}
          user: ${{ secrets.KESTRA_USER }}
          password: ${{ secrets.KESTRA_PASSWORD }}

# Trigger a flow execution via API
curl -X POST \
  http://localhost:8081/api/v1/executions/company.team/my-workflow \
  -H "Content-Type: application/json" \
  -d '{"key": "value"}'
 
# List all executions
curl http://localhost:8081/api/v1/executions?namespace=company.team
 
# Get execution details
curl http://localhost:8081/api/v1/executions/{executionId}
 
# Pause a running execution
curl -X POST \
  http://localhost:8081/api/v1/executions/{executionId}/pause

id: rag-pipeline
namespace: ai.production
 
tasks:
  # Step 1 — Scrape new documents
  - id: fetch-documents
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # Fetch new docs from knowledge base
      docs = fetch_new_documents()
 
  # Step 2 — Chunk + Embed (parallel)
  - id: embed-parallel
    type: io.kestra.plugin.core.flow.EachParallel
    value: "{{ outputs['fetch-documents'].vars.doc_ids }}"
    tasks:
      - id: embed-doc
        type: io.kestra.plugin.scripts.python.Script
        script: |
          from openai import OpenAI
          client = OpenAI()
          embedding = client.embeddings.create(...)
 
  # Step 3 — Store in vector DB
  - id: store-embeddings
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # Store in Pinecone / Qdrant / pgvector
      store_embeddings(embeddings)
 
  # Step 4 — Evaluate retrieval quality
  - id: evaluate-rag
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # Run RAGAS evaluation metrics
      score = evaluate_retrieval()
      print(f"RAG score: {score}")