ข้อมูลเบื้องต้นเกี่ยวกับวิศวกรรมข้อมูล (Introduction to Data Engineering)

ทฤษฎีและกลไกการทำงาน (How it works): Data Lakehouse คือสถาปัตยกรรมที่นำเอาความคุ้มค่าของการเก็บข้อมูลราคาประหยัดบน Object Storage (เช่น AWS S3, MinIO) มารวมเข้ากับความสามารถในการทำ ACID Transaction และการสร้างดัชนีของ Data Warehouse โดยมีเลเยอร์ Metadata หรือ Open Table Format (เช่น Apache Iceberg) ทำหน้าที่เป็นตัวเชื่อมโยงกลไกภายใน ภายใต้ฝากระโปรงของ Apache Iceberg จะมีโครงสร้างการจัดการ 3 ชั้นหลัก ได้แก่: 1. **Iceberg Catalog**: คอยชี้ตำแหน่งไฟล์ Metadata ล่าสุด (เช่น JDBC Catalog, Hive Metastore) 2. **Metadata Files**: บันทึกสกีมา (Schema), การแบ่งพาร์ทิชัน (Partition Specs) และรายการสแนปช็อต (Snapshots) ทั้งหมดของตาราง 3. **Manifest Lists & Manifest Files**: ทำหน้าที่จัดเก็บความสัมพันธ์ระดับกลุ่มข้อมูลเพื่อคัดกรองไฟล์ Parquet ที่ต้องอ่านโดยไม่ต้องกวาดอ่านข้อมูลทั้งหมด (Data File Pruning)

# Spark configuration template to catalog Iceberg with MinIO Object Storage
from pyspark.sql import SparkSession

spark = SparkSession.builder     .appName("LakehouseIcebergMinIO")     .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")     .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")     .config("spark.sql.catalog.local.type", "hadoop")     .config("spark.sql.catalog.local.warehouse", "s3a://lakehouse-warehouse/")     .config("spark.hadoop.fs.s3a.endpoint", "http://minio:9000")     .config("spark.hadoop.fs.s3a.access.key", "minioadmin")     .config("spark.hadoop.fs.s3a.secret.key", "minioadmin")     .config("spark.hadoop.fs.s3a.path.style.access", "true")     .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")     .getOrCreate()
                    

Use Case ในชีวิตจริง (Real-world Scenario): Netflix ประสบปัญหาระบบคลังข้อมูลเดิม (Hive-based Data Lake) ทำงานช้าเมื่อจัดเก็บตารางล็อกการรับชมความถี่สูง จึงสร้าง Apache Iceberg ขึ้นเพื่อทำระบบสแนปช็อต ทำให้สามารถอ่านข้อมูลแบบ Time Travel ย้อนดูประวัติสถิติย้อนหลัง และเขียนบันทึกพร้อมกันได้โดยไม่มีปัญหา Write-Conflict

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ปัญหา **Small File Problem** เกิดจากงานเขียนข้อมูลขนาดเล็กบ่อย ๆ จนไฟล์ Parquet ย่อยบวมเต็มโฟลเดอร์ ส่งผลให้ I/O อ่านช้าลงอย่างมหาศาล แก้ไขโดยการตั้งระบบทำ **Auto-Compaction** ด้วยการรันสปาร์กจ๊อบเพื่อจัดกลุ่มไฟล์ใหม่: `CALL local.system.rewrite_data_files(table => 'db.events', options => map('max-file-size-bytes', '536870912'))` (ยุบรวมให้ได้ขนาด 512MB ต่อไฟล์)

ทฤษฎีและกลไกการทำงาน (How it works): ฐานข้อมูล OLTP (เช่น PostgreSQL, MySQL) จัดเก็บข้อมูลแบบรายแถว (Row-oriented / Slotted Page) ออกแบบมาสำหรับการเข้าถึงและปรับปรุงข้อมูลทีละเรเบียนอย่างรวดเร็ว ทำงานร่วมกับดัชนี B-Tree และระบบบันทึก Log การเขียนก่อนลุยจริง (Write-Ahead Log - WAL) เพื่อการันตี ACID ขณะที่ฐานข้อมูล OLAP (เช่น ClickHouse, Snowflake, BigQuery) เก็บข้อมูลเชิงคอลัมน์ (Column-oriented) ข้อมูลของคอลัมน์เดียวกันจะถูกจัดกลุ่มและจัดเก็บเรียงลำดับอยู่ด้วยกันบนดิสก์ ช่วยให้คอมพิวเตอร์สแกนหาผลรวมของฟิลด์ตัวเลขล้านแถวได้โดยไม่ต้องโหลดคอลัมน์ที่ไม่เกี่ยวเข้ามาในแรม

-- [OLTP - PostgreSQL DDL]
CREATE TABLE customer_orders (
    order_id UUID PRIMARY KEY,
    customer_id UUID NOT NULL,
    order_amount DECIMAL(10,2) NOT NULL,
    created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW()
);
CREATE INDEX idx_customer ON customer_orders(customer_id);

-- [OLAP - ClickHouse MergeTree DDL]
CREATE TABLE analytics.customer_orders (
    order_id UUID,
    customer_id UUID,
    order_amount Decimal(10,2),
    created_at DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(created_at)
ORDER BY (customer_id, created_at);
                    

Use Case ในชีวิตจริง (Real-world Scenario): ระบบธนาคารจัดเก็บยอดชำระเงินของลูกค้าลงในระบบฐานข้อมูล Relational OLTP เพื่อความปลอดภัยในการทำรายการเงิน จากนั้นมีบริการ CDC (Change Data Capture) ส่งข้อมูลธุรกรรมเหล่านั้นไปประกอบสถิติการทุจริตในระบบ OLAP อย่าง ClickHouse เพื่อออกรายงานแดชบอร์ดความเร็วสูง

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การรันคำสั่ง `UPDATE` หรือ `DELETE` ถี่ ๆ บนระบบ OLAP จะส่งผลกระทบต่อความเร็วระบบเนื่องจากสถาปัตยกรรมจัดเก็บเชิงคอลัมน์ไม่ได้ออกแบบมารับงานนี้ แก้ไขโดยดีไซน์ตารางเป็นแบบ **Append-Only** (เน้นการใส่ข้อมูลต่อท้าย) และใช้คำสั่ง `SUM` หรือการ Join ร่วมกับวิวจดจำสถานะ (Materialized Views) ในการคำนวณหายอดปัจจุบัน

ทฤษฎีและกลไกการทำงาน (How it works): การดึงนำข้อมูล (Ingestion) เป็นการย้ายข้อมูลจากระบบต้นทางมาจัดเก็บในดิสก์หลัก โดยแบ่งเป็นสองประเภทตาม Latency: - **Batch Ingestion**: ทำงานเป็นรอบเวลา (เช่น รายชั่วโมง/รายวัน) คิวรีอ่านข้อมูลจากฐานต้นทางโดยการระบุช่วงเวลาพาร์ทิชัน เช่น `WHERE update_date >= CURRENT_DATE - INTERVAL '1 day'` ช่วยรักษาทรัพยากรคอมพิวเตอร์และคุ้มทุนคอส - **Streaming Ingestion**: ข้อมูลจะไหลต่อเนื่องทันทีที่เกิดเหตุการณ์ (Events) ผ่านระบบ Message Queue เช่น Apache Kafka ข้อมูลจะถูกเก็บในรูปแบบคิวบันทึกขนาดยาว (Distributed Commit Log) โดนแบ่งพาร์ทิชันและชี้ตำแหน่งอ่านด้วยออฟเซต (Offset) เพื่อความแม่นยำในการรันซ้ำ

# PySpark Structured Streaming configuration example from Kafka broker
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType

schema = StructType().add("order_id", StringType()).add("status", StringType())

kafka_stream = spark.readStream     .format("kafka")     .option("kafka.bootstrap.servers", "kafka-broker:9092")     .option("subscribe", "orders-topic")     .option("startingOffsets", "latest")     .load()

parsed_stream = kafka_stream     .selectExpr("CAST(value AS STRING) as json_payload")     .select(from_json(col("json_payload"), schema).alias("data"))     .select("data.*")
                    

Use Case ในชีวิตจริง (Real-world Scenario): แพลตฟอร์มเรียกรถโดยสาร (Grab/Uber) ใช้ Streaming Ingestion ดึงค่าพิกัด GPS ของผู้ขับเรียลไทม์เพื่อคำนวณราคาตามระยะทางจริง ส่วนข้อมูลสรุปบัญชีรอบวันจะใช้ระบบ Batch ในการดึงรวบรวมช่วงเที่ยงคืนเพื่อประหยัดคอมพิวต์ในการคิดเงิน

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ปัญหา **Consumer Lag** ในระบบ Streaming คือการประมวลผลข้อมูลไม่ทันปริมาณการเขียนจนเกิดการสะสมค้าง แก้ไขโดยตั้งระบบตรวจวัดผ่าน Prometheus และทำการขยายพาร์ทิชันของ Kafka (Scale Out) เพิ่มจำนวนโปรเซสผู้ดึงข้อมูลตามสัดส่วนที่เหมาะสม

ทฤษฎีและกลไกการทำงาน (How it works): Data Contract คือข้อตกลงและเงื่อนไขโครงสร้างข้อมูลรวมถึงค่าจำกัด (SLA) ระหว่างผู้ผลิตต้นทาง (Software Engineer) และผู้บริโภคข้อมูลปลายทาง (Data Engineer) โดยการนำเอากรอบการตรวจวัดประเภทข้อมูล (Schema Validation) มาคุมที่เลเยอร์การส่งออก ข้อมูลที่ไม่ผ่านข้อกำหนดจะถูกปฏิเสธตั้งแต่ต้นสาย เพื่อป้องการเกิดเหตุการณ์ฐานข้อมูลล่มเนื่องจากทีมระบบเปลี่ยนสกีมาแบบกะทันหัน

# Example of Data Contract Configuration YAML Schema
dataset: user_events
version: 1.0.0
owner: auth_team
schema:
  type: object
  properties:
    event_id: { type: string, format: uuid }
    user_id: { type: integer }
    email: { type: string, format: email }
    action: { type: string, enum: [signup, login, logout] }
  required: [event_id, user_id, email, action]
  additionalProperties: false
                    

Use Case ในชีวิตจริง (Real-world Scenario): ระบบชำระเงินของสถาบันการเงินใช้ Data Contract คุมการส่ง API บัญชีของลูกค้า หากทีมพัฒนาซอฟต์แวร์เปลี่ยนชนิดข้อมูล ID จากตัวเลขเป็นตัวหนังสือ ตัว Validate จะกั้นไม่ให้ส่งข้อมูลขึ้น Pipeline และส่งสคริปต์แจ้งเตือนกลับทีมโปรแกรมเมอร์ทันที

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การจำกัดที่เข้มงวดเกินไปอาจทำให้ระบบหลักหยุดทำงานเมื่อต้องการอัปเกรดระบบ แก้ไขโดยการทำระบบสกีมาแบบยืดหยุ่น (Schema Compatibility) เช่น การเพิ่มฟิลด์ทางเลือกได้ (Optional Fields) และสร้างสปาร์กบัฟเฟอร์แยกข้อมูลผิดปกติออกไปจัดใน Dead Letter Queue (DLQ) เพื่อสืบค้นภายหลังโดยไม่บล็อกท่อส่ง