Data Quality & Validation

Data yang buruk menyebabkan keputusan yang buruk. Sebagai data engineer, memastikan kualitas data sama pentingnya dengan membangun pipeline itu sendiri.

Dimensi Data Quality

• Completeness — Apakah semua data yang diharapkan ada? (missing values, missing rows)
• Accuracy — Apakah nilai data benar? (amount negatif, tanggal di masa depan)
• Consistency — Apakah data konsisten antar sumber? (total di source = total di warehouse)
• Timeliness — Apakah data tersedia tepat waktu?
• Uniqueness — Apakah ada duplikasi?

Great Expectations

Library Python paling populer untuk data validation. Mendefinisikan "expectation" yang harus dipenuhi data.

import great_expectations as gx

# Setup
context = gx.get_context()

# Definisi expectation suite
validator = context.sources.pandas_default.read_csv("orders.csv")

# Kolom harus ada
validator.expect_table_columns_to_match_ordered_list(
    ["order_id", "user_id", "amount", "status", "created_at"]
)

# Tidak boleh null
validator.expect_column_values_to_not_be_null("order_id")
validator.expect_column_values_to_not_be_null("amount")

# Range valid
validator.expect_column_values_to_be_between("amount", min_value=0, max_value=100000000)

# Unique
validator.expect_column_values_to_be_unique("order_id")

# Set membership
validator.expect_column_values_to_be_in_set(
    "status", ["pending", "paid", "shipped", "cancelled"]
)

# Row count
validator.expect_table_row_count_to_be_between(min_value=100, max_value=1000000)

# Run validasi
results = validator.validate()
print(f"Success: {results.success}")  # True/False

Data Contracts

Perjanjian formal antara producer dan consumer data. Mendefinisikan schema, SLA, dan quality rules.

# data_contract.yaml
schema:
  name: orders
  version: 2
  owner: backend-team

fields:
  - name: order_id
    type: integer
    required: true
    unique: true
  - name: amount
    type: decimal
    required: true
    constraints:
      min: 0
  - name: status
    type: string
    enum: [pending, paid, shipped, cancelled]
  - name: created_at
    type: timestamp
    required: true

sla:
  freshness: 1 hour
  completeness: 99.9%

alerts:
  - channel: "#data-alerts"
    on: [schema_change, quality_failure, freshness_violation]

Validation di Pipeline

def run_pipeline():
    raw_data = extract()

    # Validate setelah extract
    validate_source_data(raw_data)

    transformed = transform(raw_data)

    # Validate setelah transform
    validate_transformed(transformed)

    load(transformed)

    # Validate setelah load (reconciliation)
    source_count = len(raw_data)
    dest_count = get_warehouse_count()
    assert abs(source_count - dest_count) < 5, "Row count mismatch!"