A complete ETL (Extract, Transform, Load) pipeline demonstrating data processing workflows on Render.

Process customer signup data from CSV files with validation, cleaning, and statistical analysis. This pattern is common for:

• Data migration and imports
• Batch data processing
• Data quality monitoring
• Customer onboarding pipelines
• Report generation

• Subtask Execution: Demonstrates running tasks from other tasks with ctx.run
• Extract: Read data from CSV files (extensible to APIs, databases)
• Transform: Validate records with comprehensive error tracking
• Load: Compute statistics and prepare aggregated insights
• Error Handling: Track valid and invalid records separately
• Retry Logic: Automatic retries for transient failures
• Batch Processing: Process records individually with status tracking

run_etl_pipeline (main orchestrator)
  ├── extract_csv_data (reads CSV file)
  ├── transform_batch (validates all records)
  │   └── validate_record (called for each record)
  └── compute_statistics (aggregates results)

• Python 3.10+

# Navigate to example directory
cd etl-job

# Install dependencies
pip install -r requirements.txt

# Run the workflow service
python main.py

Service Type: Workflow

Build Command:

cd etl-job && pip install -r requirements.txt

Start Command:

cd etl-job && python main.py

Required:

• RENDER_API_KEY - Your Render API key (from Render dashboard)

1. Create Workflow Service

   • Go to Render Dashboard
   • Click "New +" → "Workflow"
   • Connect your repository
   • Name: etl-job-workflows

2. Configure Build Settings

   • Build Command: cd etl-job && pip install -r requirements.txt
   • Start Command: cd etl-job && python main.py

3. Set Environment Variables

   • Add RENDER_API_KEY in the Environment section
   • Get API key from: Render Dashboard → Account Settings → API Keys

4. Deploy

   • Click "Create Workflow"
   • Render will build and start your workflow service

Once deployed, you can test tasks directly in the Render Dashboard without writing any code:

1. Go to your Workflow service in Render Dashboard
2. Click the "Manual Run" or "Start Task" button
3. Select the task you want to test
4. Enter the task input as JSON in the text area
5. Click "Start task"

Important: The ETL pipeline expects a simple string input (the file path), not a JSON object.

Recommended Starting Point: Start with run_etl_pipeline - this is the main orchestrator task that demonstrates the complete ETL workflow (extract → transform → load).

Test the main ETL pipeline:

Task: run_etl_pipeline

Input:

"sample_data.csv"

Note: The Render Dashboard will show you the task execution status, logs, and results in real-time.

Once deployed, trigger the ETL pipeline via the Render API or SDK:

from render import Render

# Uses RENDER_API_KEY environment variable automatically
render = Render()

# Run the ETL pipeline
task_run = await render.workflows.run_task(
    "etl-job-workflows/run_etl_pipeline",
    {"source_file": "sample_data.csv"}
)

# Wait for completion
result = await task_run
print(f"Pipeline status: {result.results['status']}")
print(f"Valid records: {result.results['transform']['valid_count']}")

The example includes sample_data.csv with test data containing:

• Valid records (proper format)
• Invalid records (missing fields, bad email formats, invalid ages)

This demonstrates how the pipeline handles data quality issues.

extract_csv_data: Reads CSV file and returns records as list of dictionaries. Includes retry logic for file system issues.

validate_record: Validates a single record:

• Checks required fields (name, email)
• Validates email format
• Validates age range (0-120)
• Returns cleaned data with error tracking

transform_batch: Processes all records by running validate_record as a subtask for each one:

for record in records:
    # Run validate_record as a subtask on its own compute
    validated = await ctx.run(validate_record, record)

This demonstrates running subtasks in a loop for batch processing.

compute_statistics: Aggregates valid records to produce:

• Country distribution
• Age statistics (min, max, average)
• Data quality metrics

run_etl_pipeline: Main orchestrator that runs three subtasks sequentially:

1. await ctx.run(extract_csv_data, source_file) - Extract data
2. await ctx.run(transform_batch, raw_records) - Validate records (which runs validate_record for each)
3. await ctx.run(compute_statistics, valid_records) - Generate insights

This demonstrates sequential subtask orchestration for multi-stage pipelines.

Add Database Loading:

@app.task
async def load_to_database(ctx: TaskContext, records: list[dict]) -> dict:
    # Connect to database
    # Insert records
    # Return confirmation
    pass

Add API Data Source:

@app.task
async def extract_from_api(ctx: TaskContext, api_url: str) -> list[dict]:
    # Fetch from REST API
    # Parse JSON response
    # Return records
    pass

Add Parallel Processing:

import asyncio

@app.task
async def transform_batch_parallel(ctx: TaskContext, records: list[dict]) -> dict:
    # Validate all records in parallel
    tasks = [ctx.run(validate_record, record) for record in records]
    results = await asyncio.gather(*tasks)
    # Aggregate results
    return results

• Python-only: Workflows are only supported in Python via render
• No Blueprint Support: Workflows don't support render.yaml blueprint configuration
• Service Type: Deploy as a Workflow service on Render (not Background Worker or Web Service)