Skip to main content

Complete Pipeline

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

This pipeline combines all timestamp normalization techniques from this tutorial:

  • Format detection - Auto-detect timestamp format (ISO8601, Unix, custom)
  • Timezone conversion - Convert timestamps to UTC while preserving fractional seconds
  • Derived fields - Add hour, day_of_week, is_business_hours
  • Quality scoring - Record parsing, timestamp-presence, and UTC-format checks

Full Configuration​

Complete pipeline
normalize-timestamps.yaml
name: normalize-timestamps-complete
type: pipeline
description: Complete Timestamp Normalization Pipeline with format detection, UTC conversion, and metadata enrichment.
namespace: production
labels:
  category: data-transformation
  pattern: timestamp-normalization

config:
  input:
    kafka:
      addresses: ['${KAFKA_BROKERS:localhost:9092}']
      topics: ['${INPUT_TOPIC:raw-events}']
      consumer_group: 'timestamp-normalizer'
      start_from_oldest: false

  pipeline:
    processors:
      # Step 1: Preserve original and add metadata
      - mapping: |
          root = this
          root.timestamp_original = this.timestamp
          root.processing_metadata = {
            "pipeline_version": "v1.3.0",
            "processing_start": now().ts_format("2006-01-02T15:04:05Z07:00"),
            "stage": "format_parsing"
          }

      # Step 2: Detect timestamp format
      - mapping: |
          let ts = this.timestamp.string()

          root = this
          root.format_detected = if $ts.re_match("^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}") {
            if $ts.contains("Z") || $ts.re_match("[-+]\\d{2}:?\\d{2}$") {
              "iso8601_offset"
            } else {
              "iso8601_naive"
            }
          } else if $ts.re_match("^\\d{10}$") {
            "unix_seconds"
          } else if $ts.re_match("^\\d{13}$") {
            "unix_milliseconds"
          } else if $ts.re_match("^\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}(\\.\\d+)?$") {
            "custom_datetime"
          } else {
            "unknown"
          }

      # Step 3: Parse timestamp
      - try:
          - mapping: |
              let format = this.format_detected
              let ts = this.timestamp_original
              let local_zone = this.timezone.or("UTC")
              let local_offset = if $format == "custom_datetime" { $ts.ts_parse("2006-01-02 15:04:05").ts_format("-07:00", $local_zone) } else { null }
              let local_candidate = if $local_offset != null { ($ts + " " + $local_offset).ts_parse("2006-01-02 15:04:05 -07:00") } else { null }
              let local_resolved = if $local_candidate != null { ($ts + " " + $local_candidate.ts_format("-07:00", $local_zone)).ts_parse("2006-01-02 15:04:05 -07:00") } else { null }

              root = this

              root.timestamp_parsed = if $format == "iso8601_offset" {
                $ts.ts_parse("2006-01-02T15:04:05Z07:00")
              } else if $format == "iso8601_naive" {
                ($ts.string() + "Z").ts_parse("2006-01-02T15:04:05Z07:00")
              } else if $format == "unix_seconds" {
                $ts.number()
              } else if $format == "unix_milliseconds" {
                (($ts.number() / 1000).floor().ts_format("2006-01-02T15:04:05", "UTC") + "." + ("000" + ($ts.number() % 1000).string()).slice(-3) + "Z").ts_parse("2006-01-02T15:04:05.000Z07:00")
              } else if $format == "custom_datetime" {
                if $local_resolved.ts_format("2006-01-02 15:04:05.999999999", $local_zone) == $ts.ts_parse("2006-01-02 15:04:05").ts_format("2006-01-02 15:04:05.999999999") {
                    $local_resolved
                  } else {
                    throw("Invalid local timestamp for timezone: " + $local_zone)
                  }
              } else {
                throw("Unsupported timestamp format: " + $format)
              }

              root.processing_metadata.parse_success = true

      - catch:
          - mapping: |
              root = this
              root.processing_metadata.parse_success = false
              root.processing_metadata.parse_error = error()
              root.timestamp_parsed = null

      # Step 4: Convert to UTC
      - mapping: |
          root = this
          root.processing_metadata.stage = "timezone_conversion"

          root.timestamp = if this.processing_metadata.parse_success {
            this.timestamp_parsed.ts_format("2006-01-02T15:04:05.999999999Z07:00", "UTC")
          } else {
            this.timestamp
          }
          root.processing_metadata.conversion_success = if this.processing_metadata.parse_success { true }

      # Step 5: Enrich time metadata
      - mapping: |
          root = this
          root.processing_metadata.stage = "metadata_enrichment"

          let parsed = this.processing_metadata.parse_success
          let ts = if $parsed { this.timestamp.ts_parse("2006-01-02T15:04:05Z07:00") } else { now() }
          let year = $ts.ts_format("2006", "UTC").number()
          let month = $ts.ts_format("01", "UTC").number()
          let day = $ts.ts_format("02", "UTC").number()
          let hour = $ts.ts_format("15", "UTC").number()
          let dow = $ts.ts_strftime("%u", "UTC").number()

          root.time_metadata = if $parsed {
            {
              "year": $year,
              "month": $month,
              "day": $day,
              "hour": $hour,
              "day_of_week": $ts.ts_format("Monday", "UTC"),
              "is_weekend": $dow >= 6,
              "is_business_hours": ($dow >= 1 && $dow <= 5) && ($hour >= 9 && $hour < 17),
              "quarter": if $month <= 3 { "Q1" }
                        else if $month <= 6 { "Q2" }
                        else if $month <= 9 { "Q3" }
                        else { "Q4" },
              "hour_bucket": $ts.ts_format("2006-01-02T15", "UTC") + ":00:00Z",
              "day_bucket": $ts.ts_format("2006-01-02", "UTC")
            }
          }
          root.processing_metadata.enrichment_success = if $parsed { true }

      # Step 6: Quality scoring
      - mapping: |
          root = this

          root.validation = {
            "parse_success": this.processing_metadata.parse_success,
            "timestamp_exists": this.timestamp.type() == "string",
            "utc_format": this.timestamp.string().contains("Z")
          }

          let validations = root.validation.values()
          let true_count = $validations.map_each(v -> if v { 1 } else { 0 }).sum()
          root.quality_score = $true_count / $validations.length()

      # Step 7: Clean output
      - mapping: |
          root = this.without("processing_metadata").without("validation")

  output:
    kafka:
      addresses: ['${KAFKA_BROKERS:localhost:9092}']
      topic: '${OUTPUT_TOPIC:normalized-timestamps}'
      key: '${! json("event_id") }'
      compression: 'snappy'

logger:
  level: '${LOG_LEVEL:INFO}'
  format: json

metrics:
  prometheus:
    path: '/metrics'

Input boundary​

The displayed configuration consumes a Kafka topic selected by INPUT_TOPIC (default raw-events). The separate downloadable configuration enforces an age policy and routes invalid records to a dead-letter destination. Review its limits before choosing fixture dates or adapting its input.

Deploy​

# Deploy to Expanso orchestrator
expanso-cli job deploy normalize-timestamps.yaml

# Or run locally with expanso-edge
expanso-edge run --config normalize-timestamps.yaml

Download​

Download normalize-timestamps.yaml

What's Next?​