Data Governance คืออะไรและทำไมองค์กรต้องมี
Data Governance เป็นกรอบการทำงาน (framework) ที่กำหนดนโยบาย กระบวนการ มาตรฐาน และบทบาทหน้าที่ในการจัดการข้อมูลขององค์กร เป้าหมายหลักคือทำให้ข้อมูลมีคุณภาพ ปลอดภัย สอดคล้องกับกฎหมาย และสามารถนำไปใช้ประโยชน์ได้อย่างมีประสิทธิภาพ
เหตุผลที่องค์กรต้องมี Data Governance คือ Regulatory Compliance ที่ต้องปฏิบัติตาม PDPA, GDPR, CCPA, Data Quality ที่ข้อมูลคุณภาพต่ำทำให้ตัดสินใจผิดพลาด, Data Security ที่ป้องกันข้อมูลรั่วไหล, Cost Reduction ที่ลดค่าใช้จ่ายจากข้อมูลซ้ำซ้อนและไม่มีมาตรฐาน และ Business Intelligence ที่ต้องมีข้อมูลที่น่าเชื่อถือสำหรับการวิเคราะห์
Data Governance ไม่ใช่แค่เรื่องของ IT แต่เป็นเรื่องของทั้งองค์กร ต้องมี executive sponsorship จาก C-level, Data Stewards จากแต่ละแผนก, IT team ที่ดูแลเครื่องมือ และ Legal/Compliance team ที่ดูแลนโยบาย ทำงานร่วมกันเป็นระบบ
Framework ที่ใช้กันแพร่หลายได้แก่ DAMA-DMBOK (Data Management Body of Knowledge) ที่เป็นมาตรฐานสากล, DCAM (Data Management Capability Assessment Model) สำหรับสถาบันการเงิน และ custom frameworks ที่ออกแบบตามความต้องการขององค์กร
โครงสร้างและบทบาทใน Data Governance Framework
โครงสร้างองค์กรและบทบาทหน้าที่ที่จำเป็น
# Data Governance Organization Structure
#
# === Executive Level ===
# Chief Data Officer (CDO)
# - รับผิดชอบ data strategy ทั้งองค์กร
# - Report ตรงต่อ CEO/Board
# - กำหนดนโยบายและงบประมาณ
#
# Data Governance Council
# - ประกอบด้วย senior leaders จากทุก business unit
# - ประชุมรายเดือน
# - อนุมัตินโยบายและ resolve conflicts
# - กำหนด data priorities
#
# === Management Level ===
# Data Governance Manager
# - บริหารโปรแกรม data governance วันต่อวัน
# - ประสานงานระหว่าง business และ IT
# - ติดตาม metrics และ KPIs
#
# Data Domain Owners
# - รับผิดชอบ data domain เฉพาะ (Customer, Product, Finance)
# - ตัดสินใจเรื่อง data definitions และ business rules
# - อนุมัติ data access requests
#
# === Operational Level ===
# Data Stewards (Business)
# - ดูแลคุณภาพข้อมูลในแต่ละ domain
# - กำหนด business rules และ data standards
# - ตรวจสอบ data quality issues
#
# Data Stewards (Technical)
# - Implement data quality rules ในระบบ
# - ดูแล data pipelines และ ETL
# - จัดการ metadata และ data catalog
#
# Data Custodians (IT)
# - ดูแล infrastructure และ security
# - Backup และ disaster recovery
# - Access control และ encryption
#
# === Governance Pillars ===
# 1. Data Quality: accuracy, completeness, consistency, timeliness
# 2. Data Security: access control, encryption, masking
# 3. Data Privacy: PDPA/GDPR compliance, consent management
# 4. Metadata Management: data catalog, lineage, glossary
# 5. Data Architecture: standards, integration patterns
# 6. Data Lifecycle: creation, storage, archival, deletion
#
# === KPIs สำหรับ Data Governance ===
# - Data Quality Score: เป้า > 95%
# - Policy Compliance Rate: เป้า > 98%
# - Data Issue Resolution Time: เป้า < 48 ชั่วโมง
# - Data Catalog Coverage: เป้า > 90%
# - Access Request Processing Time: เป้า < 24 ชั่วโมง
# - Data Breach Incidents: เป้า = 0สร้าง Data Catalog ด้วย Python และ SQLite
สร้างระบบ Data Catalog สำหรับจัดการ metadata
#!/usr/bin/env python3
# data_catalog.py — Simple Data Catalog System
import sqlite3
import json
from datetime import datetime
from typing import Optional
class DataCatalog:
def __init__(self, db_path="data_catalog.db"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._create_tables()
def _create_tables(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS datasets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE NOT NULL,
domain TEXT NOT NULL,
description TEXT,
owner TEXT NOT NULL,
steward TEXT,
source_system TEXT,
format TEXT,
location TEXT,
classification TEXT DEFAULT 'internal',
pii_flag BOOLEAN DEFAULT 0,
retention_days INTEGER DEFAULT 365,
created_at TEXT DEFAULT (datetime('now')),
updated_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS columns_metadata (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dataset_id INTEGER REFERENCES datasets(id),
column_name TEXT NOT NULL,
data_type TEXT,
description TEXT,
is_nullable BOOLEAN DEFAULT 1,
is_pii BOOLEAN DEFAULT 0,
business_rule TEXT,
example_value TEXT
);
CREATE TABLE IF NOT EXISTS data_quality_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dataset_id INTEGER REFERENCES datasets(id),
column_name TEXT,
rule_type TEXT,
rule_expression TEXT,
severity TEXT DEFAULT 'warning',
is_active BOOLEAN DEFAULT 1
);
CREATE TABLE IF NOT EXISTS lineage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_dataset_id INTEGER REFERENCES datasets(id),
target_dataset_id INTEGER REFERENCES datasets(id),
transformation TEXT,
pipeline_name TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS access_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dataset_id INTEGER REFERENCES datasets(id),
user_id TEXT NOT NULL,
action TEXT NOT NULL,
timestamp TEXT DEFAULT (datetime('now')),
approved_by TEXT
);
""")
self.conn.commit()
def register_dataset(self, name, domain, owner, **kwargs):
cols = ["name", "domain", "owner"] + list(kwargs.keys())
vals = [name, domain, owner] + list(kwargs.values())
placeholders = ",".join(["?"] * len(vals))
col_names = ",".join(cols)
self.conn.execute(
f"INSERT OR REPLACE INTO datasets ({col_names}) VALUES ({placeholders})",
vals
)
self.conn.commit()
return self.conn.execute("SELECT last_insert_rowid()").fetchone()[0]
def add_column_metadata(self, dataset_name, columns):
ds = self.conn.execute(
"SELECT id FROM datasets WHERE name=?", (dataset_name,)
).fetchone()
if not ds:
raise ValueError(f"Dataset '{dataset_name}' not found")
for col in columns:
self.conn.execute("""
INSERT INTO columns_metadata
(dataset_id, column_name, data_type, description, is_pii)
VALUES (?, ?, ?, ?, ?)
""", (ds["id"], col["name"], col["type"], col.get("desc", ""), col.get("pii", False)))
self.conn.commit()
def search(self, query):
results = self.conn.execute("""
SELECT d.*, GROUP_CONCAT(c.column_name) as columns
FROM datasets d
LEFT JOIN columns_metadata c ON d.id = c.dataset_id
WHERE d.name LIKE ? OR d.description LIKE ? OR d.domain LIKE ?
GROUP BY d.id
""", (f"%{query}%", f"%{query}%", f"%{query}%")).fetchall()
return [dict(r) for r in results]
def get_lineage(self, dataset_name):
ds = self.conn.execute("SELECT id FROM datasets WHERE name=?", (dataset_name,)).fetchone()
if not ds:
return {"upstream": [], "downstream": []}
upstream = self.conn.execute("""
SELECT s.name as source, l.transformation, l.pipeline_name
FROM lineage l JOIN datasets s ON l.source_dataset_id = s.id
WHERE l.target_dataset_id = ?
""", (ds["id"],)).fetchall()
downstream = self.conn.execute("""
SELECT t.name as target, l.transformation, l.pipeline_name
FROM lineage l JOIN datasets t ON l.target_dataset_id = t.id
WHERE l.source_dataset_id = ?
""", (ds["id"],)).fetchall()
return {
"upstream": [dict(r) for r in upstream],
"downstream": [dict(r) for r in downstream],
}
# ใช้งาน
catalog = DataCatalog()
catalog.register_dataset(
name="customers", domain="Customer",
owner="CRM Team", steward="John",
description="Master customer data",
source_system="Salesforce", format="parquet",
location="s3://datalake/customers/",
classification="confidential", pii_flag=True
)
catalog.add_column_metadata("customers", [
{"name": "customer_id", "type": "INTEGER", "desc": "Unique customer ID"},
{"name": "email", "type": "VARCHAR", "desc": "Customer email", "pii": True},
{"name": "phone", "type": "VARCHAR", "desc": "Phone number", "pii": True},
{"name": "created_at", "type": "TIMESTAMP", "desc": "Registration date"},
])
results = catalog.search("customer")
print(f"Found {len(results)} datasets")Data Quality Monitoring Pipeline
สร้างระบบตรวจสอบคุณภาพข้อมูลอัตโนมัติ
#!/usr/bin/env python3
# data_quality.py — Data Quality Monitoring
import pandas as pd
import numpy as np
from datetime import datetime
from dataclasses import dataclass, field
from typing import Callable
@dataclass
class QualityRule:
name: str
column: str
check_fn: Callable
severity: str = "warning"
description: str = ""
@dataclass
class QualityResult:
rule_name: str
column: str
passed: bool
severity: str
total_rows: int
failed_rows: int
pass_rate: float
details: str = ""
class DataQualityMonitor:
def __init__(self):
self.rules: list[QualityRule] = []
self.results: list[QualityResult] = []
def add_rule(self, name, column, check_fn, severity="warning", description=""):
self.rules.append(QualityRule(name, column, check_fn, severity, description))
def add_not_null(self, column, severity="error"):
self.add_rule(f"not_null_{column}", column,
lambda df, col: df[col].notna(), severity,
f"{column} must not be null")
def add_unique(self, column, severity="error"):
self.add_rule(f"unique_{column}", column,
lambda df, col: ~df[col].duplicated(keep=False), severity,
f"{column} must be unique")
def add_range(self, column, min_val=None, max_val=None, severity="warning"):
def check(df, col):
mask = pd.Series(True, index=df.index)
if min_val is not None:
mask &= df[col] >= min_val
if max_val is not None:
mask &= df[col] <= max_val
return mask
self.add_rule(f"range_{column}", column, check, severity,
f"{column} must be between {min_val} and {max_val}")
def add_regex(self, column, pattern, severity="warning"):
self.add_rule(f"regex_{column}", column,
lambda df, col: df[col].astype(str).str.match(pattern), severity,
f"{column} must match pattern {pattern}")
def add_referential(self, column, ref_values, severity="error"):
self.add_rule(f"ref_{column}", column,
lambda df, col: df[col].isin(ref_values), severity,
f"{column} must be in reference list")
def run(self, df: pd.DataFrame) -> list[QualityResult]:
self.results = []
for rule in self.rules:
if rule.column not in df.columns:
self.results.append(QualityResult(
rule.name, rule.column, False, rule.severity,
len(df), len(df), 0.0, f"Column '{rule.column}' not found"))
continue
try:
mask = rule.check_fn(df, rule.column)
passed_count = mask.sum()
failed_count = len(df) - passed_count
pass_rate = passed_count / len(df) * 100
threshold = 100.0 if rule.severity == "error" else 95.0
self.results.append(QualityResult(
rule.name, rule.column,
pass_rate >= threshold, rule.severity,
len(df), int(failed_count), round(pass_rate, 2),
rule.description
))
except Exception as e:
self.results.append(QualityResult(
rule.name, rule.column, False, rule.severity,
len(df), len(df), 0.0, f"Error: {str(e)}"))
return self.results
def report(self):
print(f"\n{'='*60}")
print(f"Data Quality Report — {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f"{'='*60}")
passed = sum(1 for r in self.results if r.passed)
failed = len(self.results) - passed
for r in self.results:
status = "PASS" if r.passed else "FAIL"
icon = "[OK]" if r.passed else "[!!]"
print(f" {icon} {r.rule_name:30s} | {r.pass_rate:6.2f}% | {r.severity:7s} | {status}")
print(f"\nSummary: {passed} passed, {failed} failed out of {len(self.results)} rules")
score = passed / len(self.results) * 100 if self.results else 0
print(f"Overall Quality Score: {score:.1f}%")
return score
# ใช้งาน
monitor = DataQualityMonitor()
monitor.add_not_null("customer_id", severity="error")
monitor.add_not_null("email", severity="error")
monitor.add_unique("customer_id", severity="error")
monitor.add_regex("email", r"^[\w.+-]+@[\w-]+\.[\w.]+$", severity="warning")
monitor.add_range("age", min_val=0, max_val=150, severity="warning")
monitor.add_referential("status", ["active", "inactive", "suspended"], severity="error")
df = pd.DataFrame({
"customer_id": [1, 2, 3, 4, 5],
"email": ["a@b.com", "c@d.com", None, "invalid", "e@f.com"],
"age": [25, 30, -1, 200, 45],
"status": ["active", "inactive", "active", "deleted", "active"],
})
monitor.run(df)
monitor.report()Data Lineage Tracking และ Metadata Management
ติดตาม data lineage และจัดการ metadata
#!/usr/bin/env python3
# data_lineage.py — Data Lineage Tracker
import json
from datetime import datetime
from collections import defaultdict
class LineageTracker:
def __init__(self):
self.nodes = {}
self.edges = []
def register_source(self, name, system, schema=None):
self.nodes[name] = {
"type": "source", "system": system,
"schema": schema or {}, "created": datetime.now().isoformat()
}
def register_transformation(self, name, sources, targets, logic, pipeline=""):
self.nodes[name] = {
"type": "transformation", "logic": logic,
"pipeline": pipeline, "created": datetime.now().isoformat()
}
for src in sources:
self.edges.append({"from": src, "to": name, "type": "input"})
for tgt in targets:
self.edges.append({"from": name, "to": tgt, "type": "output"})
def register_dataset(self, name, storage, schema=None, classification="internal"):
self.nodes[name] = {
"type": "dataset", "storage": storage,
"schema": schema or {}, "classification": classification,
"created": datetime.now().isoformat()
}
def get_upstream(self, node_name, depth=10):
if depth <= 0:
return []
upstream = []
for edge in self.edges:
if edge["to"] == node_name:
upstream.append(edge["from"])
upstream.extend(self.get_upstream(edge["from"], depth - 1))
return list(set(upstream))
def get_downstream(self, node_name, depth=10):
if depth <= 0:
return []
downstream = []
for edge in self.edges:
if edge["from"] == node_name:
downstream.append(edge["to"])
downstream.extend(self.get_downstream(edge["to"], depth - 1))
return list(set(downstream))
def impact_analysis(self, node_name):
downstream = self.get_downstream(node_name)
affected = []
for name in downstream:
node = self.nodes.get(name, {})
affected.append({
"name": name, "type": node.get("type"),
"classification": node.get("classification", "N/A"),
})
return {"source": node_name, "affected_nodes": affected, "total": len(affected)}
def export_mermaid(self):
lines = ["graph LR"]
for edge in self.edges:
src = edge["from"].replace(" ", "_")
tgt = edge["to"].replace(" ", "_")
lines.append(f" {src} --> {tgt}")
return "\n".join(lines)
# ใช้งาน
tracker = LineageTracker()
tracker.register_source("salesforce_api", "Salesforce")
tracker.register_source("mysql_orders", "MySQL Production")
tracker.register_source("clickstream_kafka", "Kafka")
tracker.register_dataset("raw_customers", "s3://lake/raw/customers/", classification="confidential")
tracker.register_dataset("raw_orders", "s3://lake/raw/orders/", classification="internal")
tracker.register_dataset("dm_customer_360", "s3://lake/dm/customer360/", classification="confidential")
tracker.register_transformation(
"etl_customers", ["salesforce_api"], ["raw_customers"],
"Extract from Salesforce API, deduplicate, load to S3", "airflow_dag_customers"
)
tracker.register_transformation(
"etl_orders", ["mysql_orders"], ["raw_orders"],
"CDC from MySQL, transform, load to S3", "airflow_dag_orders"
)
tracker.register_transformation(
"build_customer_360", ["raw_customers", "raw_orders", "clickstream_kafka"], ["dm_customer_360"],
"Join customer + orders + clickstream, calculate LTV", "airflow_dag_c360"
)
impact = tracker.impact_analysis("salesforce_api")
print(f"Impact of changing salesforce_api: {impact['total']} nodes affected")
print(tracker.export_mermaid())Policy Enforcement และ Access Control
สร้างระบบ policy enforcement สำหรับ data access
#!/usr/bin/env python3
# data_policy.py — Data Governance Policy Engine
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
class Classification(Enum):
PUBLIC = "public"
INTERNAL = "internal"
CONFIDENTIAL = "confidential"
RESTRICTED = "restricted"
class Action(Enum):
READ = "read"
WRITE = "write"
DELETE = "delete"
EXPORT = "export"
SHARE = "share"
@dataclass
class Policy:
name: str
classification: Classification
allowed_roles: list[str]
allowed_actions: list[Action]
requires_approval: bool = False
max_rows_export: Optional[int] = None
data_masking: bool = False
audit_required: bool = True
class PolicyEngine:
def __init__(self):
self.policies: list[Policy] = []
self.audit_log: list[dict] = []
self._setup_default_policies()
def _setup_default_policies(self):
self.policies = [
Policy("public_read", Classification.PUBLIC,
["viewer","analyst","engineer","admin"],
[Action.READ, Action.EXPORT], max_rows_export=1000000),
Policy("internal_read", Classification.INTERNAL,
["analyst","engineer","admin"],
[Action.READ, Action.EXPORT], max_rows_export=100000),
Policy("internal_write", Classification.INTERNAL,
["engineer","admin"],
[Action.WRITE]),
Policy("confidential_read", Classification.CONFIDENTIAL,
["analyst","engineer","admin"],
[Action.READ], data_masking=True),
Policy("confidential_export", Classification.CONFIDENTIAL,
["admin"],
[Action.EXPORT], requires_approval=True, max_rows_export=10000),
Policy("restricted_read", Classification.RESTRICTED,
["admin"],
[Action.READ], requires_approval=True, data_masking=True),
Policy("restricted_all", Classification.RESTRICTED,
["admin"],
[Action.WRITE, Action.DELETE, Action.EXPORT, Action.SHARE],
requires_approval=True),
]
def check_access(self, user_role, classification, action, dataset_name=""):
matching = [p for p in self.policies
if p.classification == classification
and action in p.allowed_actions
and user_role in p.allowed_roles]
if not matching:
self._audit("DENIED", user_role, classification, action, dataset_name)
return {"allowed": False, "reason": "No matching policy"}
policy = matching[0]
result = {
"allowed": True,
"policy": policy.name,
"requires_approval": policy.requires_approval,
"data_masking": policy.data_masking,
"max_rows": policy.max_rows_export,
}
self._audit("ALLOWED", user_role, classification, action, dataset_name, policy.name)
return result
def _audit(self, decision, role, classification, action, dataset, policy=""):
self.audit_log.append({
"timestamp": datetime.now().isoformat(),
"decision": decision, "role": role,
"classification": classification.value,
"action": action.value, "dataset": dataset,
"policy": policy,
})
def apply_masking(self, df, pii_columns):
masked = df.copy()
for col in pii_columns:
if col in masked.columns:
masked[col] = masked[col].astype(str).apply(
lambda x: x[:2] + "*" * (len(x)-4) + x[-2:] if len(x) > 4 else "****"
)
return masked
# ใช้งาน
engine = PolicyEngine()
result = engine.check_access("analyst", Classification.CONFIDENTIAL, Action.READ, "customers")
print(f"Access: {result['allowed']}, Masking: {result['data_masking']}")
result = engine.check_access("viewer", Classification.RESTRICTED, Action.READ, "salary_data")
print(f"Access: {result['allowed']}, Reason: {result.get('reason')}")FAQ คำถามที่พบบ่อย
Q: Data Governance กับ Data Management ต่างกันอย่างไร?
A: Data Governance เป็นเรื่องของนโยบาย กฎ และการตัดสินใจว่า data ควรถูกจัดการอย่างไร (what and why) ส่วน Data Management เป็นเรื่องของการปฏิบัติจริงตามนโยบาย (how) เช่น การสร้าง pipelines, ETL, storage, backup ทั้งสองทำงานร่วมกัน Governance กำหนดทิศทาง Management ลงมือทำ
Q: องค์กรขนาดเล็กต้องมี Data Governance ไหม?
A: ทุกองค์กรที่จัดการข้อมูลควรมี Data Governance ในระดับที่เหมาะสม องค์กรเล็กอาจไม่ต้องมีทีมเต็มรูปแบบ แต่ควรมีอย่างน้อย data owner ที่รับผิดชอบข้อมูลแต่ละชุด นโยบายพื้นฐานเรื่อง access control และ data quality checks เบื้องต้น เริ่มจากสิ่งที่จำเป็นแล้วค่อยขยาย
Q: เครื่องมือ Data Governance ที่แนะนำมีอะไรบ้าง?
A: Open source ได้แก่ Apache Atlas สำหรับ metadata management, Great Expectations สำหรับ data quality, OpenMetadata สำหรับ data catalog และ Amundsen จาก Lyft สำหรับ data discovery Commercial ได้แก่ Collibra, Alation, Informatica, Atlan แต่ละตัวเหมาะกับขนาดและความต้องการขององค์กรที่ต่างกัน
Q: PDPA กับ Data Governance เกี่ยวข้องกันอย่างไร?
A: PDPA (พ. ร. บ. คุ้มครองข้อมูลส่วนบุคคล) กำหนดให้องค์กรต้องจัดการข้อมูลส่วนบุคคลอย่างเหมาะสม Data Governance framework ช่วยปฏิบัติตาม PDPA ผ่าน data classification ที่ระบุว่าข้อมูลไหนเป็น PII, access control ที่จำกัดการเข้าถึง, data lineage ที่ติดตามว่าข้อมูลไปที่ไหน, consent management ที่จัดการความยินยอม และ data retention policy ที่กำหนดระยะเวลาเก็บข้อมูล
