Step 6: Validate Required Fields

Reject records with missing required fields before they reach BigQuery. This replaces DataStage Data Rules stages with conditional validation.

The Goal

• Check that transaction_id, customer_id, and amount_usd are present
• Reject invalid records with a clear error message
• Prevent bad data from polluting your BigQuery tables

Why This Matters

Data Quality: Bad records in BigQuery break dashboards and reports.

Delivery boundary: Reject invalid records before the BigQuery output; measure storage and query billing separately.

Debugging: Clear error messages help identify source data issues.

DataStage Equivalent

In DataStage, validation typically requires:

1. Data Rules Stage or Filter Stage
2. Reject links to capture bad records
3. Separate error handling jobs

Mark invalid records and route them through an output switch.

Implementation

Add validation after steps 1–5 and replace the output with this switch:

Partial snippet

step-6-validate.yaml

pipeline:
  processors:
    - switch:
        - check: this.transaction_id == null || this.customer_id == null || this.amount_usd == null
          processors:
            - mapping: |
                root = this
                root._error = "Missing required field for BigQuery load"
                root._failed_at = now()

output:
  switch:
    cases:
      - check: errored() || this._error != null
        output:
          file:
            path: etl-dlq.jsonl
            codec: lines
      - check: '!errored() && this._error == null'
        output:
          gcp_bigquery:
            project: '${GCP_PROJECT}'
            dataset: financial_data
            table: transactions
            format: NEWLINE_DELIMITED_JSON
            write_disposition: WRITE_APPEND
            # Partition the destination table by _partition_date when creating it in BigQuery.
            batching:
              count: 1000
              period: 30s

Understanding the Code

Expression	What It Does
this.field == null	Check for a missing or null field
root._error	Mark a rejected record with a reason
errored()	Route earlier processor failures to the DLQ
output.switch	Deliver each record to the DLQ or BigQuery

Expected Behavior

Valid Record: Reaches the BigQuery output unchanged

Invalid Record (missing customer_id):

The file DLQ receives the rejected record with _error and _failed_at. The BigQuery output does not receive it.

Production Considerations

Detailed Error Messages

Identify which field is missing:

Partial snippet

- mapping: |
    let missing = []
    let missing = if this.transaction_id == null { $missing.append("transaction_id") } else { $missing }
    let missing = if this.customer_id == null { $missing.append("customer_id") } else { $missing }
    let missing = if this.amount_usd == null { $missing.append("amount_usd") } else { $missing }

    root = if $missing.length() > 0 {
      throw("Missing required fields: " + $missing.join(", "))
    } else {
      this
    }

Dead Letter Queue

The output switch above writes rejected records to etl-dlq.jsonl. Set that path to durable storage on the Edge node. Records with an error flag or a processor error never enter the BigQuery case.

Type Validation

Check data types, not just presence:

Partial snippet

- mapping: |
    root = if this.amount_usd.type() != "number" {
      throw("amount_usd must be numeric, got: " + this.amount_usd.type())
    } else if this.amount_usd < 0 {
      throw("amount_usd cannot be negative: " + this.amount_usd.string())
    } else {
      this
    }

Range Validation

Flag suspicious values:

Partial snippet

- mapping: |
    root = this
    root._warnings = []

    # Flag unusually large transactions
    root._warnings = if this.amount_usd > 100000 {
      root._warnings.append("Large transaction: review required")
    } else {
      root._warnings
    }

    # Flag future dates
    root._warnings = if this.transaction_date > now().ts_format("2006-01-02") {
      root._warnings.append("Future transaction date")
    } else {
      root._warnings
    }

Validation Metrics

Count validation failures for monitoring:

Partial snippet

- metric:
    type: counter
    name: validation_failures
    labels:
      reason: ${! meta("error_reason").or("unknown") }

---

Complete Pipeline

You've built all 6 transformation steps! See the complete, complete example configuration:

📖 Complete Pipeline →