Skip to main content

Complete Pipeline

Copy pipeline YAML
Review it for your environment before you deploy.

This pipeline combines all deduplication techniques from this tutorial:

  • Hash-based deduplication - Use content hash to detect exact duplicates
  • Fingerprint deduplication - Detect near-duplicates with key field matching
  • ID-based deduplication - Use event_id for idempotent processing
  • Time-windowed dedup - Limit cache comparison to the configured time window

Full Configuration​

Complete pipeline
deduplicate-events.yaml
name: deduplicate-events-complete
type: pipeline
description: Complete Event Deduplication Pipeline - Detects and filters duplicate events using cache-based deduplication.
namespace: production
labels:
  category: data-transformation
  pattern: deduplication

config:
  cache_resources:
    - label: dedup_cache
      memory:
        default_ttl: 1h

  input:
    http_server:
      address: '0.0.0.0:8080'
      path: /events
      allowed_verbs: ['POST']

  pipeline:
    threads: 1
    processors:
      # Step 1: Parse and validate
      - mapping: root = content().parse_json()
      - mapping: |
          root = this
          root.received_at = now()

          # Validate required fields
          root = if !this.exists("event_id") {
            throw("Missing required field: event_id")
          } else {
            this
          }

      # Step 2: Generate deduplication key
      - mapping: |
          root = this

          # Create composite key for deduplication
          # Options: event_id only, or event_id + type + timestamp window
          # Composite: event_type + source + 1-minute window
          root.dedup_key = if this.dedup_strategy.or("event_id") == "composite" {
            this.event_type.or("unknown") + ":" + this.source.or("unknown") + ":" + (this.timestamp.or(now()).ts_unix() / 60).floor().string()
          } else {
            # Simple: just event_id
            this.event_id
          }

          meta dedup_key = root.dedup_key

      - group_by_value:
          value: '${! this.dedup_key }'
      - mapping: |
          root = this
          meta batch_duplicate = batch_index() > 0

      # Steps 3-5 run serially per message so duplicate keys in the same input
      # batch cannot race between the cache lookup and write.
      - for_each:
          - branch:
              request_map: root = this.dedup_key
              processors:
                - cache:
                    resource: dedup_cache
                    operator: exists
                    key: '${! content() }'
              result_map: |
                root.is_duplicate = meta("batch_duplicate").bool() || content().string().bool()
                meta cache_result = content().string()
                meta is_duplicate = root.is_duplicate
          - catch:
              - mapping: |
                  root = this
                  root.is_duplicate = false
                  meta cache_result = ""
                  meta is_duplicate = false
          - mapping: |
              root = this
              root.dedup_result = {
                "is_duplicate": this.is_duplicate,
                "dedup_key": meta("dedup_key"),
                "checked_at": now()
              }
              root.duplicate_metadata = if this.is_duplicate {
                {
                  "detected_at": now(),
                  "original_hash": this.dedup_key,
                  "strategy": this.dedup_strategy.or("event_id")
                }
              } else { deleted() }
          - switch:
              - check: '!this.is_duplicate'
                processors:
                  - cache:
                      resource: dedup_cache
                      operator: set
                      key: '${! meta("dedup_key") }'
                      value: '${! json("received_at") }'

  output:
    switch:
      cases:
        - check: '!this.is_duplicate'
          output:
            kafka:
              addresses: ['${KAFKA_BROKERS:localhost:9092}']
              topic: unique-events
              batching:
                count: 100
                period: 5s
        - output:
            file:
              path: "/var/expanso/duplicates/${!timestamp_unix_date('2006-01-02')}/dupes.jsonl"
              codec: lines

logger:
  level: INFO
  format: json

metrics:
  prometheus:
    path: /metrics

Quick Test​

# Send an event
curl -X POST http://localhost:8080/events \
  -H "Content-Type: application/json" \
  -d '{"event_id": "evt_001", "data": "hello"}'

# Send the same event again (will be deduplicated)
curl -X POST http://localhost:8080/events \
  -H "Content-Type: application/json" \
  -d '{"event_id": "evt_001", "data": "hello"}'

# Intended result: the repeated key is dropped

Deploy​

# Deploy to Expanso orchestrator
expanso-cli job deploy deduplicate-events.yaml

# Or run locally with expanso-edge
expanso-edge run --config deduplicate-events.yaml

Download​

Download deduplicate-events.yaml

What's Next?​