Data Generation & Transformation

This example demonstrates using the generate input to create synthetic data for testing pipelines. It's useful for development, testing transformations, and simulating data sources without external dependencies.

Download & Run

Quick Start:

# 1. Download the pipeline and check it offline
curl -sSLo pipeline.yaml \
  https://docs.expanso.io/examples/data-transformer.yaml
expanso-edge validate pipeline.yaml

# 2. Start a local edge agent (its state stays in ./.edge)
expanso-edge run --local --data-dir ./.edge

# 3. In a second terminal, deploy the pipeline to it as a job
export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli"
{ printf 'name: data-transformer\ntype: pipeline\nconfig:\n'
  sed 's/^/  /' pipeline.yaml; } |
  expanso-cli --endpoint http://localhost:9010 job deploy -

This runs on your machine with no account. Step 3 wraps the pipeline in the name, type and config fields that a job needs, and a stdout output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See Local Mode for more, or Deploy to Cloud to run it on managed nodes.

Download: data-transformer.yaml

What This Pipeline Does

Generates messages every 5 seconds with synthetic data
Creates unique IDs, timestamps, and counter values
Enriches with processing metadata
Outputs to stdout for inspection

Complete Pipeline

input:
  generate:
    interval: 5s
    mapping: |
      root.id = uuid_v4()
      root.message = "Hello from Expanso"
      root.timestamp = now()
      root.count = counter()

pipeline:
  processors:
    - mapping: |
        root = this
        root.processed_at = now()

output:
  stdout:
    codec: lines

Try Locally

This pipeline needs nothing external, so it runs on a local agent with no account. Start one as in Step 3 of Try It Without an Account, then run this in a second terminal, from the same folder:

curl -fsSL -o data-transformer.yaml \
  https://docs.expanso.io/examples/data-transformer.yaml
expanso-edge validate data-transformer.yaml
{ printf 'name: data-transformer\ntype: pipeline\nconfig:\n'
  sed 's/^/  /' data-transformer.yaml; } > data-transformer-job.yaml
export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli"
expanso-cli job deploy data-transformer-job.yaml \
  --endpoint http://localhost:9010

The printf and sed lines wrap the pipeline in the job fields (name, type, config) that job deploy expects. EXPANSO_CLI_CONFIG_DIR and --endpoint keep the CLI on the local agent, away from any saved profile or Expanso Cloud workspace. Every 5 seconds a message like this appears in the agent's terminal (IDs and times vary):

{"count":1,"id":"...","message":"Hello from Expanso","processed_at":"...","timestamp":"..."}

The job keeps running until you stop it:

expanso-cli job stop data-transformer --force \
  --endpoint http://localhost:9010

On a local agent the pipeline can keep producing output for a short while after the stop (22 seconds in one test), then halts. Stopping the agent with Ctrl+C stops everything at once.

Configuration Breakdown

Input: Generate

input:
  generate:
    interval: 5s
    mapping: |
      root.id = uuid_v4()
      root.message = "Hello from Expanso"
      root.timestamp = now()
      root.count = counter()

The generate input creates messages at a fixed interval using a Bloblang mapping. Each message gets:

id: Unique UUID (e.g., 550e8400-e29b-41d4-a716-446655440000)
message: Static string
timestamp: Current timestamp in RFC3339 format
count: Incrementing counter starting from 1

See: Generate Input

Processor: Add Metadata

pipeline:
  processors:
    - mapping: |
        root = this
        root.processed_at = now()

Copies the entire input message and adds a processed_at timestamp showing when the message was processed.

Output: Stdout

output:
  stdout:
    codec: lines

Prints each message to stdout as a single line of JSON.

See: Stdout Output

Example Output

Each message will look like this:

{"id":"a1b2c3d4-e5f6-7890-abcd-ef1234567890","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:00Z","count":1,"processed_at":"2024-01-15T10:30:00.123Z"}
{"id":"b2c3d4e5-f6a7-8901-bcde-f12345678901","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:05Z","count":2,"processed_at":"2024-01-15T10:30:05.456Z"}
{"id":"c3d4e5f6-a7b8-9012-cdef-012345678902","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:10Z","count":3,"processed_at":"2024-01-15T10:30:10.789Z"}

Common Variations

Generate Realistic Test Data

Create more complex synthetic data:

input:
  generate:
    interval: 1s
    mapping: |
      root.user_id = "user-" + counter().string()
      root.event_type = ["login", "logout", "purchase", "view"].index(counter() % 4)
      root.timestamp = now()
      root.value = random_int(min: 1, max: 100)
      root.metadata.ip = "192.168.1." + (random_int(min: 1, max: 254)).string()
      root.metadata.user_agent = "TestAgent/1.0"

Generate Batch Test Data

Generate a specific number of messages as fast as possible:

input:
  generate:
    interval: ""  # Empty interval = as fast as possible
    count: 1000   # Generate exactly 1000 messages
    mapping: |
      root.id = uuid_v4()
      root.batch_num = counter()
      root.data = "test-data-" + counter().string()

Scheduled Processing

Use generate to trigger scheduled processing (e.g., every 5 minutes):

input:
  generate:
    interval: '@every 5m'
    mapping: 'root = {}'

pipeline:
  processors:
    # Add your scheduled processing here
    - mapping: |
        root.scheduled_run = now()
        root.task = "periodic-cleanup"

Simulate Streaming Data

Generate continuous events with varying data:

input:
  generate:
    interval: 100ms  # 10 messages per second
    mapping: |
      root.sensor_id = "sensor-" + (random_int(min: 1, max: 10)).string()
      root.temperature = random_int(min: 15, max: 30)
      root.humidity = random_int(min: 30, max: 80)
      root.timestamp = now()

Transform Generated Data

Apply complex transformations to synthetic data:

input:
  generate:
    interval: 2s
    mapping: |
      root.raw_value = random_int(min: 0, max: 1000)
      root.timestamp = now()

pipeline:
  processors:
    - mapping: |
        # Calculate derived values
        root.value = this.raw_value
        root.category = if this.raw_value < 300 {
          "low"
        } else if this.raw_value < 700 {
          "medium"
        } else {
          "high"
        }
        root.normalized = this.raw_value / 1000.0
        root.processed_at = now()

Use Cases

Development & Testing

Generate test data without external dependencies:

Test transformations before connecting to real data sources
Validate pipeline configurations
Benchmark processing performance
Debug routing logic

Scheduled Tasks

Trigger periodic processing:

Run cleanup tasks every hour
Fetch data from APIs on schedule
Generate periodic reports
Trigger batch jobs

Load Testing

Simulate high-volume data sources:

Test pipeline throughput
Validate error handling under load
Measure resource usage
Identify bottlenecks

Data Simulation

Create realistic test scenarios:

Simulate sensor data for IoT testing
Generate user events for analytics testing
Create transaction data for fraud detection testing
Mock API responses

Important Notes

Interval Formats

The interval field supports several formats:

interval: 1s          # Every second
interval: 500ms       # Every 500 milliseconds
interval: 1m          # Every minute
interval: '@every 5m' # Every 5 minutes (cron-style)
interval: ""          # As fast as possible (use with count)

Counter Behavior

The counter() function maintains state across messages:

Starts at 1
Increments by 1 for each message
Persists across the lifetime of the input
Resets if the pipeline restarts

Performance Considerations

For high-volume generation:

Use shorter intervals (e.g., 100ms for 10 msgs/sec)
Set count to limit total messages
Use batch_size to group messages
Monitor resource usage

Memory Management

When generating large volumes:

Set explicit count to prevent infinite generation
Use appropriate interval to control rate
Consider downstream processing capacity
Monitor memory usage

Next Steps

Generate Input - Full configuration options
Mapping Processor - Transform data with Bloblang
Bloblang Guide - Data transformation reference

Download & Run​

What This Pipeline Does​

Complete Pipeline​

Try Locally​

Configuration Breakdown​

Input: Generate​

Processor: Add Metadata​

Output: Stdout​

Example Output​

Common Variations​

Generate Realistic Test Data​

Generate Batch Test Data​

Scheduled Processing​

Simulate Streaming Data​

Transform Generated Data​

Use Cases​

Development & Testing​

Scheduled Tasks​

Load Testing​

Data Simulation​

Important Notes​

Interval Formats​

Counter Behavior​

Performance Considerations​

Memory Management​

Next Steps​