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

Tool	Language	เหมาะกับ	Cloud	ราคา
Soda	YAML (SodaCL)	ทุกู้คืน	Soda Cloud	Free + Paid
Great Expectations	Python	Developer	GX Cloud	Free + Paid
dbt Tests	SQL/YAML	Analytics Engineer	dbt Cloud	Free + Paid
Monte Carlo	No-Code	Data Team	SaaS Only	$$$$
Elementary	SQL	dbt Users	Self-hosted	Free

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

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

ข้อดี	ข้อเสีย
ประสิทธิภาพสูง ทำงานได้เร็วและแม่นยำ ลดเวลาทำงานซ้ำซ้อน	ต้องใช้เวลาเรียนรู้เบื้องต้นพอสมควร มี Learning Curve สูง
มี Community ขนาดใหญ่ มีคนช่วยเหลือและแหล่งเรียนรู้มากมาย	บางฟีเจอร์อาจยังไม่เสถียร หรือมีการเปลี่ยนแปลงบ่อยในเวอร์ชันใหม่
รองรับ Integration กับเครื่องมือและบริการอื่นได้หลากหลาย	ต้นทุนอาจสูงสำหรับ Enterprise License หรือ Cloud Service
เป็น Open Source หรือมีเวอร์ชันฟรีให้เริ่มต้นใช้งาน	ต้องการ Hardware หรือ Infrastructure ที่เพียงพอ

จากตารางเปรียบเทียบจะเห็นว่าข้อดีมีมากกว่าข้อเสียอย่างชัดเจน โดยเฉพาะในแง่ของประสิทธิภาพและความสามารถในการ 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