Soda Data Quality Citizen Developer

Soda Data Quality

Soda Data Quality SodaCL YAML Citizen Developer Data Contracts Automated Testing Missing Duplicates Schema Freshness Anomaly PostgreSQL BigQuery Snowflake CI/CD

SodaCL Checks

# === Soda Setup & SodaCL ===

# pip install soda-core-postgres
# pip install soda-core-bigquery

# configuration.yml
# data_source my_postgres:
#   type: postgres
#   host: localhost
#   port: 5432
#   username: 
#   password: 
#   database: analytics
#   schema: public

# checks.yml — SodaCL Data Quality Checks
# checks for orders:
#   # Row count
#   - row_count > 0
#   - row_count between 1000 and 100000
#
#   # Missing values
#   - missing_count(customer_id) = 0
#   - missing_count(email) = 0
#   - missing_count(total_amount) = 0
#
#   # Duplicates
#   - duplicate_count(order_id) = 0
#
#   # Freshness
#   - freshness(created_at) < 1d
#
#   # Valid values
#   - invalid_count(status) = 0:
#       valid values: ["pending", "confirmed", "shipped", "delivered", "cancelled"]
#
#   # Schema check
#   - schema:
#       fail:
#         when required column missing: [order_id, customer_id, total_amount]
#         when wrong column type:
#           order_id: integer
#           total_amount: numeric
#
#   # Custom SQL
#   - failed rows:
#       fail query: |
#         SELECT * FROM orders
#         WHERE total_amount < 0

# Run: soda scan -d my_postgres -c configuration.yml checks.yml

from dataclasses import dataclass
from typing import List

@dataclass
class CheckResult:
    check: str
    table: str
    result: str
    value: str
    status: str

results = [
    CheckResult("row_count > 0", "orders", "PASS", "45,230", "OK"),
    CheckResult("missing_count(email)", "orders", "PASS", "0", "OK"),
    CheckResult("duplicate_count(order_id)", "orders", "PASS", "0", "OK"),
    CheckResult("freshness(created_at) < 1d", "orders", "PASS", "2h", "OK"),
    CheckResult("invalid_count(status)", "orders", "FAIL", "15", "ALERT"),
    CheckResult("row_count > 0", "products", "PASS", "8,500", "OK"),
    CheckResult("missing_count(price)", "products", "WARN", "3", "WARN"),
]

print("=== Soda Scan Results ===")
passed = sum(1 for r in results if r.status == "OK")
failed = sum(1 for r in results if r.status != "OK")
for r in results:
    print(f"  [{r.status}] {r.table}.{r.check} = {r.value}")
print(f"\n  Summary: {passed} passed, {failed} issues")

Data Contracts

# === Data Contracts with Soda ===

# data_contract_orders.yml
# dataset: orders
# version: "2.0"
# owner: data-team@example.com
# description: "Order data contract"
#
# schema:
#   required_columns:
#     - name: order_id
#       type: integer
#       not_null: true
#       unique: true
#     - name: customer_id
#       type: integer
#       not_null: true
#     - name: total_amount
#       type: numeric
#       not_null: true
#       min: 0
#     - name: status
#       type: varchar
#       valid_values: ["pending", "confirmed", "shipped", "delivered"]
#     - name: created_at
#       type: timestamp
#       not_null: true
#
# quality:
#   freshness: "< 1 hour"
#   completeness: "> 99%"
#   uniqueness:
#     order_id: "100%"
#
# sla:
#   availability: "99.9%"
#   latency: "< 15 minutes"

@dataclass
class DataContract:
    dataset: str
    owner: str
    checks: int
    freshness: str
    completeness: str
    status: str

contracts = [
    DataContract("orders", "data-team", 12, "< 1h", "99.8%", "VALID"),
    DataContract("products", "product-team", 8, "< 24h", "99.5%", "VALID"),
    DataContract("customers", "crm-team", 10, "< 6h", "98.2%", "BREACH"),
    DataContract("payments", "finance-team", 15, "< 30min", "99.9%", "VALID"),
    DataContract("inventory", "ops-team", 6, "< 1h", "97.5%", "BREACH"),
]

print("\n=== Data Contracts ===")
for c in contracts:
    print(f"  [{c.status}] {c.dataset} (Owner: {c.owner})")
    print(f"    Checks: {c.checks} | Fresh: {c.freshness} | Complete: {c.completeness}")

CI/CD Integration

# === CI/CD Pipeline Integration ===

# GitHub Actions
# .github/workflows/data-quality.yml
# name: Data Quality Checks
# on:
#   schedule:
#     - cron: '0 * * * *'  # ทุกชั่วโมง
#   workflow_dispatch:
#
# jobs:
#   soda-scan:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v4
#       - uses: actions/setup-python@v5
#         with: { python-version: '3.11' }
#       - run: pip install soda-core-postgres
#       - run: |
#           soda scan -d production \
#             -c configuration.yml \
#             checks/*.yml
#         env:
#           POSTGRES_HOST: }
#           POSTGRES_PASSWORD: }
#       - uses: slackapi/slack-github-action@v1
#         if: failure()
#         with:
#           payload: '{"text": "Data Quality Check FAILED!"}'

# dbt + Soda Integration
# dbt run && soda scan -d warehouse checks/post_dbt.yml

ci_pipeline = [
    {"step": "dbt run", "purpose": "Transform Data", "duration": "5 min"},
    {"step": "soda scan (critical)", "purpose": "Check Critical Quality", "duration": "30 sec"},
    {"step": "dbt test", "purpose": "dbt Built-in Tests", "duration": "1 min"},
    {"step": "soda scan (full)", "purpose": "All Quality Checks", "duration": "2 min"},
    {"step": "Notify Slack", "purpose": "Alert on Failure", "duration": "< 1 sec"},
]

print("CI/CD Pipeline:")
for i, step in enumerate(ci_pipeline, 1):
    print(f"  {i}. [{step['step']}] {step['purpose']} ({step['duration']})")

# Citizen Developer Workflow
workflow = {
    "Day 1": "เรียนรู้ SodaCL YAML Format 30 นาที",
    "Day 2": "เขียน Basic Checks (row_count, missing, duplicate)",
    "Day 3": "เพิ่ม Freshness และ Schema Checks",
    "Day 4": "เชื่อม Soda Cloud Dashboard + Slack Alert",
    "Day 5": "สร้าง Data Contract สำหรับ Dataset หลัก",
}

print(f"\n\nCitizen Developer Onboarding:")
for day, task in workflow.items():
    print(f"  [{day}]: {task}")

เคล็ดลับ

• Start Simple: เริ่มจาก row_count missing duplicate ก่อน
• Freshness: ตรวจ Freshness ทุก Table ป้องกัน Stale Data
• CI/CD: รัน Soda Scan หลัง dbt run ทุกครั้ง
• Contracts: สร้าง Data Contract สำหรับ Dataset สำคัญ
• Alert: แจ้ง Slack เมื่อ Check Fail อย่ารอดู Dashboard

Best Practices สำหรับนักพัฒนา

การเขียนโค้ดที่ดีไม่ใช่แค่ทำให้โปรแกรมทำงานได้ แต่ต้องเขียนให้อ่านง่าย ดูแลรักษาง่าย และ Scale ได้ หลัก SOLID Principles เป็นพื้นฐานสำคัญที่นักพัฒนาทุกู้คืนควรเข้าใจ ได้แก่ Single Responsibility ที่แต่ละ Class ทำหน้าที่เดียว Open-Closed ที่เปิดให้ขยายแต่ปิดการแก้ไข Liskov Substitution ที่ Subclass ต้องใช้แทน Parent ได้ Interface Segregation ที่แยก Interface ให้เล็ก และ Dependency Inversion ที่พึ่งพา Abstraction ไม่ใช่ Implementation

เรื่อง Testing ก็ขาดไม่ได้ ควรเขียน Unit Test ครอบคลุมอย่างน้อย 80% ของ Code Base ใช้ Integration Test ทดสอบการทำงานร่วมกันของ Module ต่างๆ และ E2E Test สำหรับ Critical User Flow เครื่องมือยอดนิยมเช่น Jest, Pytest, JUnit ช่วยให้การเขียน Test เป็นเรื่องง่าย

เรื่อง Version Control ด้วย Git ใช้ Branch Strategy ที่เหมาะกับทีม เช่น Git Flow สำหรับโปรเจคใหญ่ หรือ Trunk-Based Development สำหรับทีมที่ Deploy บ่อย ทำ Code Review ทุก Pull Request และใช้ CI/CD Pipeline ทำ Automated Testing และ Deployment

เปรียบเทียบข้อดีและข้อเสีย

จากตารางเปรียบเทียบจะเห็นว่าข้อดีมีมากกว่าข้อเสียอย่างชัดเจน โดยเฉพาะในแง่ของประสิทธิภาพและความสามารถในการ Scale สำหรับข้อเสียส่วนใหญ่สามารถแก้ไขได้ด้วยการเรียนรู้อย่างเป็นระบบและวางแผนทรัพยากรให้เหมาะสม

สรุป

Soda Data Quality SodaCL YAML Citizen Developer Data Contracts Missing Duplicates Schema Freshness Anomaly CI/CD dbt Soda Cloud Dashboard Alert PostgreSQL BigQuery Snowflake