Run a Python Script with Outputs and Metrics in Kestra

id: python-generate-outputs
namespace: company.team

variables:
  file: orders.csv

tasks:
  - id: analyze_orders
    type: io.kestra.plugin.scripts.python.Script
    dependencies:
      - faker
      - kestra
    outputFiles:
      - "*.csv"
    script: |
      import csv
      import random
      import time
      from faker import Faker
      from kestra import Kestra

      start_time = time.time()

      fake = Faker()

      # list of columns for the CSV file
      columns = [
          "order_id",
          "customer_name",
          "customer_email",
          "product_id",
          "price",
          "quantity",
          "total",
      ]

      filename = "{{ vars.file }}"
      tags = {'file': filename}

      # Generate 100 random orders
      orders = []
      for i in range(100):
          order_id = i + 1
          customer_name = fake.name()
          customer_email = fake.email()
          product_id = random.randint(1, 20)
          price = round(random.uniform(10.0, 200.0), 2)
          quantity = random.randint(1, 10)
          total = round(price * quantity, 2)
          orders.append(
              [order_id, customer_name, customer_email, product_id, price, quantity, total]
          )

      # Write the orders to a CSV file
      with open(filename, "w", newline="") as file:
          writer = csv.writer(file)
          writer.writerow(columns)
          writer.writerows(orders)

      # Calculate and print the sum and average of the "total" column
      total_sum = sum(order[6] for order in orders)
      average_order = round(total_sum / len(orders), 2)
      print(f"Total sum: {total_sum}")
      print(f"Average Order value: {average_order}")

      Kestra.outputs({"total_sum": total_sum, "average_order": average_order})
      Kestra.counter('total_sum', total_sum, tags)
      Kestra.counter('average_order', average_order, tags)

      end_time = time.time()  
      processing_time = end_time - start_time
      Kestra.timer('processing_time', processing_time, tags)

      print(f"The script execution took: {processing_time} seconds")

Run a Python script as an orchestrated task that produces real, observable results: a downloadable file, structured outputs, and custom metrics. This blueprint generates a CSV of 100 synthetic orders, computes the total and average order value, and surfaces those numbers as Kestra outputs and metrics so downstream tasks, dashboards, and stakeholders can consume them. It solves a common gap with ad hoc Python: scripts that write files and print numbers locally but leave nothing tracked, versioned, or shareable.

How it works

The analyze_orders task runs io.kestra.plugin.scripts.python.Script with its dependencies set to faker and kestra, installed automatically at runtime.
The script reads the {{ vars.file }} variable (default orders.csv) to decide the output filename, avoiding hardcoded paths.
It builds 100 fake orders with Faker, writes them to the CSV, then computes total_sum and average_order.
outputFiles is set to *.csv, so Kestra captures the generated file into internal storage for download.
The script calls Kestra.outputs(...) for total_sum and average_order, Kestra.counter(...) for both values, and Kestra.timer(...) for processing_time, all tagged with the filename.

What you get

A CSV artifact downloadable from the Execution page in the UI.
Named outputs (total_sum, average_order) usable by later tasks via expressions.
Custom counters and a timer visible in the Kestra metrics view.
A fully parametrized run controlled by a single variable.

Who it's for

Data engineers wrapping existing Python in a tracked, schedulable pipeline.
Analysts who need shareable file outputs without manual exports.
Platform teams standardizing how scripts emit metrics and artifacts.

Why orchestrate this with Kestra

Python's own runtime has no scheduler, no retry policy, and no built-in way to track outputs or metrics across runs. Kestra adds event and schedule triggers, automatic retries, execution lineage, and declarative YAML so the same script becomes reproducible and observable. Outputs and outputFiles are captured centrally instead of vanishing on the worker, and metrics are queryable over time.

Prerequisites

A running Kestra instance.
Python available in the execution environment (the task installs faker and kestra).

Secrets

None. This blueprint references no secret(...) values.

Quick start

Add the flow to your Kestra instance.
Execute it as is to generate orders.csv.
Open the Execution page to download the file and inspect outputs.
Check the metrics tab for the counters and timer.

How to extend

Override the file variable to change the output name per execution.
Add a schedule or event trigger to run it automatically.
Replace the synthetic data logic with a real extract from your source.
Pass total_sum into a downstream notification or database task.

Run a Python script and generate outputs, metrics and files specified with a variable

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

New to Kestra?

Run a Python script and generate outputs, metrics and files specified with a variable

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Share this Blueprint

Create a Blueprint

More Data Blueprints

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

Build an ETL Pipeline with Airbyte Cloud and dbt Core for SaaS Analytics

Sync Postgres records to Algolia index

Run a Scheduled ETL Sync with Airbyte Cloud

More Core Blueprints

Advanced Scheduling Automation, Run Tasks Only on Business Days by Country

Layered Error Handling with Retries and KV Context

Infrastructure Automation workflow example: Getting started with Kestra

Reduce LLM API Costs with a Redis Semantic Cache for Repeated Questions

Run a Python script in a Kubernetes pod

Fetch logs from all pods in a Kubernetes namespace

New to Kestra?