สัปดาห์ที่ 08 · Data with AI

ตาราง — list / dict / CSV / JSON / SQLite

ในคอร์สเดิม สิ่งเหล่านี้สอนกระจาย 4 หัวข้อ · ในคู่มือนี้รวมเป็นเรื่องเดียว — "ทุกอย่างคือตาราง" · แค่เก็บที่ต่างกัน เลือกตามขนาด/ความถาวร/จำนวนผู้ใช้

เป้าหมายสัปดาห์นี้

มอง list/dict/JSON/CSV/SQL เป็น "ตารางที่เก็บคนละแบบ"
อ่าน-เขียน CSV, JSON, SQLite ด้วย Python ได้
ออกแบบ schema — field, type, primary key, relationships
ใช้ Pandas เป็น "universal translator" ระหว่างทุก format
ทำ groupby / merge / filter ด้วย pandas

🔒 ทำไม "ออกแบบ schema" คือทักษะที่ AI ทำให้คุณไม่ได้ AI เขียน read_csv ได้ใน 5 วินาที · แต่ "ตัดสินใจว่าจะเก็บ field อะไร · type อะไร · จับคู่ตารางยังไง" = งานของคุณ · ถ้าออกแบบผิดตอนแรก — แก้ทีหลังเจ็บปวด · AI ก็ไม่รู้ว่า Grading App ของคุณ จะใช้กี่ user · เก็บนานแค่ไหน · ต้อง query อะไรบ้าง · นั่นเป็น domain knowledge ของคุณ

🎯 Running Case — เก็บข้อมูล Grading App ของพี่นัท

W04-W07 เราสร้าง logic ของ Grading App แล้ว · ตอนนี้ต้องตัดสินใจว่า "ข้อมูลจริง" จะเก็บไว้ที่ไหน · พี่นัทมีข้อมูล 4 ชนิด — นักศึกษา 50 คน · drawing assignments · rubric items · คะแนน · เลือก format ผิด = ปวดหัวภายหลัง

🗺 Data Storage Ladder — เลือกตามขนาดของปัญหา

ทุก format มีจุดที่ "ดีที่สุด" · ไต่ขึ้นเมื่อปัญหาใหญ่ขึ้น · ไม่ใช่ "ใช้ SQL ตลอด"

flowchart TB L1["1. ตัวแปร (RAM)
list / dict
ทดลอง · code ตัวอย่าง · ใน function
❌ ปิดโปรแกรม = หาย"] L2["2. ไฟล์ในเครื่อง
.json / .csv / .txt
script ส่วนตัว · config · < 1000 row
❌ 2 คนเขียนพร้อมกัน = ชน"] L3["3. Database 1 ไฟล์
SQLite
app เล็ก · 1 user · query ซับซ้อน
❌ หลายคนพร้อมกัน lock ทั้งไฟล์"] L4["4. Database จริงจัง
PostgreSQL / MySQL
หลาย user · production · network
⚠️ ต้อง host server · backup"] L1 -->|"> 1000 row
ต้องคงอยู่"| L2 L2 -->|"ต้อง search/filter
หรือมี relationship"| L3 L3 -->|"หลาย user/server
พร้อมกัน"| L4 classDef ram fill:#0d3f3a,stroke:#34d399,color:#fff classDef file fill:#3d2c1a,stroke:#ffd43b,color:#fff classDef sqlite fill:#1e3a5f,stroke:#3776ab,color:#fff classDef prod fill:#5c1818,stroke:#ef4444,color:#fff class L1 ram class L2 file class L3 sqlite class L4 prod

กฎ — เริ่มล่างสุดเสมอ Grading App ใหม่ → เริ่ม list/dict ใน RAM · พอเริ่มทดลองจริง → JSON file · พอใส่จริง 50 คน → SQLite · ถ้ามี TA หลายคนใช้พร้อมกัน → PostgreSQL · 90% ของ project นักศึกษาปี 1 อยู่แค่ขั้น 1-2

🏭 ข้อมูลที่ Engineer เจอจริง

ไม่ใช่แค่ list ของนักศึกษา — ลองดูว่าแต่ละสาขาเจอข้อมูลแบบไหน

สาขา	ข้อมูลที่เจอจริง	Format ที่ใช้บ่อย
Mechatronics / Robotics	Sensor log (อุณหภูมิ, vibration, distance)	CSV (จาก data logger) · MQTT → InfluxDB
Electrical / Power	Energy meter readings · Power quality	CSV ทุก 15 นาที · SQLite ใน gateway
Civil / Smart Building	Building automation log · Occupancy	CSV หรือ JSON ผ่าน BMS
Chemical	Lab measurement · Batch records	Excel/CSV · LIMS database
Industrial	Production data · Quality control	SQL (MES) · Excel report
Mechanical	Part list (BOM) · CAD properties	Excel · JSON (config)
ทุกสาขา (vibe-coding)	Form responses · API JSON · config	Google Sheets · JSON · SQLite

สังเกต — CSV ครองโลก vendor data เกือบทุก data logger / instrument / sensor export มาเป็น CSV · ถ้าทำงานกับ hardware แทบจะเห็น CSV ตลอด · ต้องอ่าน + clean CSV ให้คล่อง = essential skill

🔑 หลักคิด — All Data is a Table

ลองดู — ทุกอย่างมีหน้าตาเหมือนตาราง · แค่ภาษา/รูปแบบต่างกัน

id	name	email	year
1	Ploy	ploy@ubu.ac.th	1
2	Bank	bank@ubu.ac.th	1
3	Mint	mint@ubu.ac.th	2

เขียนแบบเดียวกัน — 5 รูปแบบ

1. Python list of dicts (in RAM):

students = [
    {"id": 1, "name": "Ploy", "email": "ploy@ubu.ac.th", "year": 1},
    {"id": 2, "name": "Bank", "email": "bank@ubu.ac.th", "year": 1},
    {"id": 3, "name": "Mint", "email": "mint@ubu.ac.th", "year": 2},
]

2. JSON file (in disk):

[
  {"id": 1, "name": "Ploy", "email": "ploy@ubu.ac.th", "year": 1},
  {"id": 2, "name": "Bank", "email": "bank@ubu.ac.th", "year": 1},
  {"id": 3, "name": "Mint", "email": "mint@ubu.ac.th", "year": 2}
]

3. CSV file (in disk):

id,name,email,year
1,Ploy,ploy@ubu.ac.th,1
2,Bank,bank@ubu.ac.th,1
3,Mint,mint@ubu.ac.th,2

4. Pandas DataFrame (RAM, but "table-aware"):

import pandas as pd
df = pd.read_csv("students.csv")
print(df)

5. SQLite (file but searchable):

CREATE TABLE students (
    id INTEGER PRIMARY KEY,
    name TEXT,
    email TEXT,
    year INTEGER
);
INSERT INTO students VALUES (1, 'Ploy', 'ploy@ubu.ac.th', 1);
...

เลือกอันไหน? ดูได้จากตารางใน W02 — Database Overview · สั้น ๆ: เริ่มจาก list/dict → เมื่อข้อมูลเยอะค่อยใส่ JSON/CSV → เมื่อต้อง query/multi-user ค่อยขึ้น SQLite

🧱 list — ลำดับของอะไรก็ได้

scores = [85, 72, 91, 60, 45] # เข้าถึงสมาชิก (index เริ่มที่ 0) print("scores[0] =", scores[0]) # ตัวแรก print("scores[-1] =", scores[-1]) # ตัวสุดท้าย print("scores[1:3] =", scores[1:3]) # slice # เพิ่ม / ลบ scores.append(78) scores.insert(0, 100) print("after append+insert:", scores) # functions ที่ใช้บ่อย print("len:", len(scores), "| sum:", sum(scores)) print("max:", max(scores), "| min:", min(scores)) print("sorted:", sorted(scores))

List comprehension — เขียน loop เป็น 1 บรรทัด

scores = [85, 72, 91, 60, 45, 33] # เดิม: 3 บรรทัด doubled = [] for s in scores: doubled.append(s * 2) print("doubled (long):", doubled) # ใหม่: 1 บรรทัด doubled2 = [s * 2 for s in scores] print("doubled (comp):", doubled2) # มี filter passing = [s for s in scores if s >= 50] print("passing:", passing) # ใช้ใน data work บ่อยมาก emails = [{"id": 1, "name": "Ploy"}, {"id": 2, "name": "Bank"}] names = [s["name"] for s in emails] print("names:", names)

nums = [1, 2, 3, 4, 5]
result = [n * n for n in nums if n % 2 == 0]
print(result)

[4, 16]

Filter n % 2 == 0 = เลขคู่ → [2, 4] · แล้ว map n * n → [4, 16]

📚 dict — คีย์ → ค่า

student = { "id": "640001", "name": "Ploy", "scores": [85, 72, 91], } # เข้าถึง print("name:", student["name"]) print("phone:", student.get("phone", "N/A")) # default ถ้าไม่มี # เพิ่ม / แก้ student["email"] = "ploy@ubu.ac.th" print("\nafter add email:") for key, value in student.items(): print(f" {key}: {value}") # nested dict — common in real data drawing = { "id": "D001", "student": {"id": "640001", "name": "Ploy"}, "scores": {"R1": 28, "R2": 18, "R3": 22, "R4": 24} } print("\nstudent name:", drawing["student"]["name"]) print("R1 score: ", drawing["scores"]["R1"]) print("total: ", sum(drawing["scores"].values()))

d = {"a": [1, 2, 3], "b": [4, 5]}
print(len(d), sum(d["a"]), d.get("c", 0))

2 6 0

len(d) = จำนวน key = 2 · sum(d["a"]) = 1+2+3 = 6 · d.get("c", 0) = ไม่มี key "c" → default 0

# ทุก key / ทุก value
list(student.keys())
list(student.values())

📦 JSON — เซฟ dict/list ลงไฟล์

เปิดไฟล์ใน Cursor: open("students.json", "w", encoding="utf-8") · ใน browser นี้ Pyodide ใช้ virtual file system จึงรันได้เลย

import json # Grading App data — เก็บนักศึกษา students = [ {"id": "640001", "name": "Ploy", "year": 2, "major": "MAE"}, {"id": "640002", "name": "Bank", "year": 2, "major": "EE"}, {"id": "640003", "name": "Mint", "year": 2, "major": "MAE"}, ] # เขียน JSON — กฎทอง: utf-8 + ensure_ascii=False สำหรับภาษาไทย with open("students.json", "w", encoding="utf-8") as f: json.dump(students, f, ensure_ascii=False, indent=2) # อ่านกลับ with open("students.json", "r", encoding="utf-8") as f: loaded = json.load(f) print("จำนวน:", len(loaded)) for s in loaded: print(f" {s['id']}: {s['name']} (year {s['year']}, {s['major']})") # ดู raw JSON ที่เขียนลงไฟล์ print("\n=== raw JSON ===") with open("students.json", encoding="utf-8") as f: print(f.read())

กฎทอง — encoding="utf-8" + ensure_ascii=False เสมอ ลองลบ ensure_ascii=False ใน runner ด้านบนแล้ว Run — ภาษาไทยจะกลายเป็น ปล.. · นี่คือสิ่งที่ทำให้ JSON ดู "ขยะ"

📊 CSV — มาตรฐานของข้อมูลจากภายนอก

import csv students = [ {"id": "640001", "name": "Ploy", "year": 2, "major": "MAE"}, {"id": "640002", "name": "Bank", "year": 2, "major": "EE"}, {"id": "640003", "name": "Mint", "year": 2, "major": "MAE"}, ] # เขียน CSV with open("students.csv", "w", encoding="utf-8", newline="") as f: fieldnames = ["id", "name", "year", "major"] writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(students) # ดู raw CSV print("=== raw CSV ===") with open("students.csv", encoding="utf-8") as f: print(f.read()) # อ่านกลับ with open("students.csv", encoding="utf-8") as f: reader = csv.DictReader(f) rows = list(reader) print(f"อ่านได้ {len(rows)} แถว") print(f"row 0: {rows[0]}") print(f"name ของ row 1: {rows[1]['name']}")

CSV จาก Excel/Google Sheets ระวัง

encoding อาจจะเป็น cp874 (Thai Windows) ไม่ใช่ utf-8 — เปิด Excel → Save As → CSV UTF-8 (Comma delimited)
Excel ชอบเปลี่ยน "001234" เป็น 1234 — เก็บ ID เป็น string ใน source data
field ที่มี comma ต้องครอบด้วย "..."
CSV ไม่มี type — อ่านทุก field มาเป็น string เสมอ · ต้อง int() เอง

🐼 Pandas — Universal Translator + Data Wrangler

Pandas คือ Excel + SQL + format converter รวมกัน · runner นี้โหลด pandas package (~20s ครั้งแรก · cache หลังจากนั้น) — แล้วคุณรัน "จริง" ได้

📥 1) อ่านได้ทุก format — เขียนได้ทุก format

นี่คือ superpower ของ Pandas — ใช้เป็น universal translator

import pandas as pd import json # เริ่มต้นจาก list of dicts students = [ {"id": "640001", "name": "Ploy", "year": 2, "major": "MAE", "gpa": 3.45}, {"id": "640002", "name": "Bank", "year": 2, "major": "EE", "gpa": 3.20}, {"id": "640003", "name": "Mint", "year": 2, "major": "MAE", "gpa": 3.78}, {"id": "640004", "name": "Tum", "year": 1, "major": "ChE", "gpa": 2.90}, ] # → DataFrame df = pd.DataFrame(students) print("=== DataFrame ===") print(df) # → CSV df.to_csv("out.csv", index=False, encoding="utf-8") # → JSON df.to_json("out.json", orient="records", force_ascii=False, indent=2) # → SQLite (built-in) import sqlite3 conn = sqlite3.connect("out.db") df.to_sql("students", conn, if_exists="replace", index=False) conn.close() # อ่านกลับจาก SQLite — แสดงว่าเขียนสำเร็จ conn = sqlite3.connect("out.db") df2 = pd.read_sql("SELECT * FROM students WHERE gpa >= 3.5", conn) print("\n=== query กลับจาก SQLite (gpa >= 3.5) ===") print(df2)

💡 ครั้งแรกใช้เวลา ~20 วินาทีโหลด pandas + sqlite3 (built-in Python) · ครั้งต่อ ๆ ไปเร็ว

👀 2) Inspect — รู้จักข้อมูลก่อนใช้งาน

import pandas as pd # Sample: drawing scores ของห้องเรียนพี่นัท data = { "student_id": ["640001", "640002", "640003", "640004", "640005"], "name": ["Ploy", "Bank", "Mint", "Tum", "Aof"], "drawing_id": ["D1", "D1", "D1", "D1", "D1"], "R1": [28, 25, 30, 20, 26], # Dimensions "R2": [18, 15, 20, 12, 18], # Line quality "R3": [22, 20, 23, 18, 21], # Annotations "R4": [24, 22, 25, 18, 24], # Title block } df = pd.DataFrame(data) print("--- head() ---") # 5 แถวแรก print(df.head()) print("\n--- shape ---") print(df.shape) # (rows, cols) print("\n--- columns ---") print(list(df.columns)) print("\n--- dtypes ---") print(df.dtypes) # type ของแต่ละ column print("\n--- describe() ---") print(df.describe()) # mean, std, min, max ของ numeric

🎯 3) Select & Filter — เลือกแถว/คอลัมน์

import pandas as pd df = pd.DataFrame({ "name": ["Ploy", "Bank", "Mint", "Tum", "Aof"], "year": [2, 2, 2, 1, 3], "gpa": [3.45, 3.20, 3.78, 2.90, 3.60], "major": ["MAE", "EE", "MAE", "ChE", "MAE"], }) # เลือก 1 column → Series print("--- df['name'] ---") print(df["name"].tolist()) # เลือกหลาย columns → DataFrame print("\n--- df[['name', 'gpa']] ---") print(df[["name", "gpa"]]) # Filter ด้วย boolean print("\n--- gpa >= 3.5 ---") print(df[df["gpa"] >= 3.5]) # Filter หลายเงื่อนไข — ใช้ & และ | print("\n--- MAE และ gpa >= 3.4 ---") print(df[(df["major"] == "MAE") & (df["gpa"] >= 3.4)]) # .loc[row, col] — เลือกตาม label print("\n--- df.loc[0, 'name'] ---") print(df.loc[0, "name"]) # .iloc[row, col] — เลือกตาม position print("\n--- df.iloc[-1] (แถวสุดท้าย) ---") print(df.iloc[-1])

📊 4) Groupby — สรุปข้อมูลตามกลุ่ม

import pandas as pd # scores ของนักศึกษา 50 คน — แต่ละคนทำ drawing 3 ชิ้น data = { "student_id": ["640001", "640001", "640001", "640002", "640002", "640002", "640003", "640003", "640003"], "drawing": ["D1", "D2", "D3"] * 3, "score": [78, 82, 85, 65, 70, 72, 90, 88, 92], } df = pd.DataFrame(data) # คะแนนเฉลี่ยของแต่ละนักศึกษา print("--- เฉลี่ยต่อคน ---") print(df.groupby("student_id")["score"].mean()) # จำนวน drawing ของแต่ละคน print("\n--- นับ drawing ต่อคน ---") print(df.groupby("student_id").size()) # หลาย aggregation พร้อมกัน print("\n--- multi agg ---") print(df.groupby("student_id")["score"].agg(["mean", "min", "max", "count"])) # Groupby + filter top = df.groupby("student_id")["score"].mean() print("\n--- ใครเฉลี่ย > 80 ---") print(top[top > 80])

🔗 5) Merge — รวมตาราง (Join)

import pandas as pd # 2 ตาราง — แยกกัน เพื่อไม่ซ้ำข้อมูล students = pd.DataFrame({ "id": ["640001", "640002", "640003"], "name": ["Ploy", "Bank", "Mint"], "year": [2, 2, 2], }) scores = pd.DataFrame({ "student_id": ["640001", "640001", "640002", "640003"], "drawing": ["D1", "D2", "D1", "D1"], "score": [85, 90, 70, 95], }) # Merge — เหมือน SQL JOIN merged = students.merge(scores, left_on="id", right_on="student_id") print("--- merged ---") print(merged) # ตอนนี้ groupby ก็ได้ชื่อแล้ว print("\n--- เฉลี่ยพร้อมชื่อ ---") print(merged.groupby("name")["score"].mean())

🧰 6) value_counts & apply — utility ที่ใช้บ่อยมาก

import pandas as pd df = pd.DataFrame({ "name": ["Ploy", "Bank", "Mint", "Tum", "Aof", "Som", "Pat"], "major": ["MAE", "EE", "MAE", "ChE", "MAE", "EE", "MAE"], "score": [85, 70, 92, 60, 78, 88, 75], }) # value_counts — นับ unique values print("--- จำนวนนักศึกษาแต่ละสาขา ---") print(df["major"].value_counts()) # apply — function กับทุกแถว def to_grade(s): if s >= 80: return "A" if s >= 70: return "B" if s >= 60: return "C" return "F" df["grade"] = df["score"].apply(to_grade) print("\n--- พร้อม grade column ---") print(df) # นับ grade print("\n--- กระจาย grade ---") print(df["grade"].value_counts())

6 moves นี้ครอบคลุม 80% ของ data work ในอาชีพ อ่าน → inspect → filter → groupby → merge → apply · ฝึกให้คล่อง — Final Project และทุกงานต่อ ๆ ไปจะใช้

💾 SQLite — Database 1 ไฟล์

เมื่อข้อมูลเริ่มเยอะ + ต้อง search/filter ซับซ้อน · CSV ช้า · ใช้ SQLite

import sqlite3 # สร้าง/เปิด database (สร้างไฟล์ถ้ายังไม่มี) conn = sqlite3.connect("grading.db") cur = conn.cursor() # สร้าง table — ระบุ type + constraints cur.execute(""" CREATE TABLE IF NOT EXISTS students ( id TEXT PRIMARY KEY, name TEXT NOT NULL, year INTEGER, major TEXT ) """) # เพิ่มแถว — ใช้ ? เพื่อ safe (ป้องกัน SQL injection) data = [ ("640001", "Ploy", 2, "MAE"), ("640002", "Bank", 2, "EE"), ("640003", "Mint", 2, "MAE"), ("640004", "Tum", 1, "ChE"), ] cur.executemany( "INSERT OR REPLACE INTO students VALUES (?, ?, ?, ?)", data ) conn.commit() # Query 1: SELECT ทั้งหมด print("--- ทั้งหมด ---") for row in cur.execute("SELECT * FROM students"): print(row) # Query 2: WHERE filter print("\n--- เฉพาะ MAE ---") for row in cur.execute("SELECT name, year FROM students WHERE major = ?", ("MAE",)): print(row) # Query 3: aggregation print("\n--- นับนักศึกษาแต่ละสาขา ---") for row in cur.execute("SELECT major, COUNT(*) FROM students GROUP BY major"): print(row) conn.close()

SQL พื้นฐาน 5 คำสั่ง

`SELECT cols FROM table`	ดึงข้อมูลออก
`INSERT INTO table VALUES (...)`	เพิ่มแถว
`UPDATE table SET col=val WHERE ...`	แก้แถว
`DELETE FROM table WHERE ...`	ลบแถว
`WHERE / GROUP BY / ORDER BY`	กรอง / จัดกลุ่ม / เรียง

SQLite ใน Pandas ก็ได้

df = pd.read_sql("SELECT * FROM students WHERE year = 1", conn)
df.to_sql("students_2026", conn, if_exists="replace")

ใช้ pandas สำหรับ analyze + SQLite สำหรับ persistence — ของ Python รับกันได้หมด

📐 Schema Design — ออกแบบ "ตาราง" ก่อนเก็บ

นี่คือทักษะ human-only ของเฟส Data · ตัดสินใจว่า ตารางไหน · field อะไร · type อะไร · ความสัมพันธ์ยังไง = ส่งผลตลอด lifecycle ของ project

🔑 4 คำถามที่ต้องตอบก่อนสร้าง schema

มี entity (สิ่งของ) อะไรบ้าง? — Grading App มี: Student · Drawing · Rubric · Score
แต่ละ entity มี field อะไร? — Student: id, name, year, major · Drawing: id, student_id, due_date, …
field ไหนเป็น Primary Key (PK)? — ID เฉพาะของแต่ละ row · ห้ามซ้ำ · เช่น student_id = "640001"
มี relationship ระหว่าง entity มั้ย? — Drawing ของใคร? → ใส่ student_id เป็น Foreign Key (FK) ใน Drawing

📊 Schema ของ Grading App (ตัวอย่างเต็ม)

erDiagram STUDENT ||--o{ DRAWING : ส่ง DRAWING ||--o{ SCORE : ได้คะแนน RUBRIC ||--o{ SCORE : ตาม_criteria STUDENT { string id PK "เช่น 640001" string name "ชื่อ-นามสกุล" int year "1-4" string major "MAE/EE/ChE..." string email } DRAWING { string id PK "D001" string student_id FK date submitted_at string filename } RUBRIC { string id PK "R1" string criteria "Dimensions accurate" int weight "0-100" } SCORE { int id PK string drawing_id FK string rubric_id FK int score "0-weight" string note }

📝 กฎ Schema Design ที่ใช้ได้กับทุก project

กฎ	เพราะ	ตัวอย่าง
ทุก table มี PK	ต้องระบุได้ว่า row ไหน "ตัวเดียวกัน"	`id TEXT PRIMARY KEY`
PK ห้ามเปลี่ยน	ถ้าเปลี่ยน → FK ทุกที่พัง	ใช้ UUID หรือ student_id ที่ไม่เปลี่ยน
1 field = 1 ข้อมูล (Atomic)	ห้ามเก็บ "Ploy, Bank, Mint" ใน 1 cell	ถ้ามีหลายค่า → ทำ table ใหม่
ห้ามทำซ้ำ (Don't Repeat)	ถ้าข้อมูลซ้ำ → update ที่เดียว ไม่ใช่หลายที่	student.name เก็บที่ STUDENT table เท่านั้น
Type ที่ตรงกับธรรมชาติของข้อมูล	ตัวเลข → INTEGER · วันที่ → DATE · ไม่ใช่ TEXT ทุกอัน	price `REAL` ไม่ใช่ `TEXT`
NOT NULL ถ้า required	ป้องกันข้อมูลขาด	`name TEXT NOT NULL`

🚩 Schema ที่ "ผิดพลาดที่พบบ่อย"

❌ เก็บหลายค่าใน column เดียว

# ผิด — เก็บคะแนน 4 criteria ในช่องเดียว
students = [
    {"id": "640001", "name": "Ploy", "scores": "28,18,22,24"},
]
# ถูก — แยกตาราง
scores = [
    {"student_id": "640001", "rubric_id": "R1", "score": 28},
    {"student_id": "640001", "rubric_id": "R2", "score": 18},
    ...
]

❌ ทำซ้ำชื่อในหลายที่

# ผิด — เก็บชื่อนักศึกษาทุกแถวของ scores
scores = [
    {"student_id": "640001", "name": "Ploy", "score": 28},
    {"student_id": "640001", "name": "Ploy", "score": 18},
    # ถ้า Ploy เปลี่ยนชื่อ → ต้องแก้ทุกแถว
]
# ถูก — เก็บชื่อใน students แล้ว merge เมื่อต้องใช้
students = [{"id": "640001", "name": "Ploy"}]
scores = [{"student_id": "640001", "score": 28}, ...]

เคล็ดลับ — ใช้ AI สร้าง schema จาก spec ลองส่ง prompt: "ออกแบบ schema สำหรับ Grading App ที่มี Student/Drawing/Rubric/Score · แสดงเป็น Mermaid erDiagram" — AI สร้างให้ได้ดี · แต่คุณต้อง "review" ด้วยกฎ 6 ข้อด้านบน

🔗 SQL Deep Dive — JOIN · Aggregate · NULL · CASE

📌 ก่อนเริ่ม — "ไม่ต้องใช้ SQL ทุกครั้ง" สำหรับ script เล็ก ๆ · CLI tool · config · 1-user app → JSON file ดีกว่า · ง่ายกว่า · อ่านได้ด้วยตา · ไม่ต้อง setup · ดู Data Storage Ladder ด้านบน · ขยับขึ้น SQL "เมื่อจำเป็น" เท่านั้น · ส่วนต่อไปนี้คือทักษะ SQL "สำหรับวันที่จำเป็น" ไม่ใช่ "ต้องใช้ตั้งแต่ Day 1"

W08 ส่วน SQLite เบื้องต้นใช้ตารางเดียว · ในความจริง "data ดี ๆ อยู่ในหลายตาราง" (ตาม Schema Design ด้านบน) · ตอนจะใช้ต้อง JOIN กลับมา · นี่คือทักษะ SQL ที่ทำเงินจริง

📊 เลือก storage ตามขนาดของ project

Project size	เลือก	เหตุผล
script ใช้ครั้งเดียว · < 10 records	list / dict ใน RAM	ไม่ต้อง persist
CLI tool · config · 1 user · < 1000 records	JSON file ✅	ง่ายที่สุด · มนุษย์อ่านได้ · gitignore ได้
Data analysis · CSV จาก vendor	CSV + pandas	มาตรฐาน · Excel เปิดได้
app เล็ก · multi-table · ต้อง query	SQLite ✅	1 file · ไม่ต้อง server
multi-user · production	PostgreSQL	concurrent · network

👇 ถ้า project ของคุณตกหมวด "JSON ก็พอ" — "skip section นี้ได้" · กลับมาอ่านเมื่อจำเป็น

🔗 1) JOIN — รวม 2 ตารางด้วย Foreign Key

ใน Grading App: students มีข้อมูลคน · scores มีคะแนน · อยากดู "ชื่อ + คะแนน" ในรายการเดียว → ต้อง JOIN

import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() # สร้าง 2 ตาราง cur.executescript(""" CREATE TABLE students ( id TEXT PRIMARY KEY, name TEXT, major TEXT ); CREATE TABLE scores ( id INTEGER PRIMARY KEY, student_id TEXT, subject TEXT, score INTEGER ); INSERT INTO students VALUES ('640001', 'Ploy', 'MAE'), ('640002', 'Bank', 'EE'), ('640003', 'Mint', 'MAE'), ('640004', 'Tum', 'ChE'); INSERT INTO scores VALUES (1, '640001', 'Drawing', 85), (2, '640001', 'Lab', 78), (3, '640002', 'Drawing', 70), (4, '640002', 'Lab', 82), (5, '640003', 'Drawing', 92); -- 640004 (Tum) ไม่มีคะแนนเลย! """) print("=== INNER JOIN — เฉพาะคนที่มีคะแนน ===") for row in cur.execute(""" SELECT s.name, sc.subject, sc.score FROM students s INNER JOIN scores sc ON s.id = sc.student_id ORDER BY s.name """): print(row) # Tum ไม่ปรากฏ — เพราะไม่มีคะแนน print("\n=== LEFT JOIN — รวม Tum ด้วย (NULL ตรงคะแนน) ===") for row in cur.execute(""" SELECT s.name, sc.subject, sc.score FROM students s LEFT JOIN scores sc ON s.id = sc.student_id ORDER BY s.name """): print(row) # Tum ปรากฏ · subject=None · score=None

JOIN type	คืนแถวเมื่อ	ใช้เมื่อ
INNER JOIN	match ทั้ง 2 ตาราง	ดูเฉพาะคนที่มี scores · default ถ้าไม่ระบุ
LEFT JOIN	ทุกแถวฝั่งซ้าย + match ฝั่งขวา (ที่ขาดเป็น NULL)	"รายชื่อทั้งหมด · ใครยังไม่ส่งงาน"
RIGHT JOIN	ตรงข้าม LEFT (SQLite ไม่มี · ใช้ swap LEFT แทน)	ไม่ค่อยใช้
FULL OUTER JOIN	ทุกแถวทั้ง 2 ตาราง	ไม่ค่อยใช้ · SQLite ไม่รองรับ

กฎทอง — เริ่มที่ INNER · ใช้ LEFT เมื่อ "อยากเห็นคนที่ขาด" 90% ของ JOIN ในงานจริงคือ INNER · LEFT ใช้เพื่อตรวจ "ใครที่ยังไม่มีข้อมูล"

🔗 JOIN 3 ตาราง — chain ได้

import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() cur.executescript(""" CREATE TABLE students (id TEXT PRIMARY KEY, name TEXT); CREATE TABLE drawings (id TEXT PRIMARY KEY, student_id TEXT, title TEXT); CREATE TABLE rubrics (id TEXT PRIMARY KEY, drawing_id TEXT, score INTEGER); INSERT INTO students VALUES ('640001', 'Ploy'), ('640002', 'Bank'); INSERT INTO drawings VALUES ('D1', '640001', 'House'), ('D2', '640001', 'Engine'), ('D3', '640002', 'Bridge'); INSERT INTO rubrics VALUES ('R1', 'D1', 28), ('R2', 'D1', 18), ('R3', 'D2', 25), ('R4', 'D3', 30); """) # JOIN 3 ตาราง — รายชื่อ + ภาพ + คะแนน print("=== Full chain JOIN ===") for row in cur.execute(""" SELECT s.name, d.title, r.score FROM students s INNER JOIN drawings d ON s.id = d.student_id INNER JOIN rubrics r ON d.id = r.drawing_id ORDER BY s.name, d.title """): print(row)

📊 2) Aggregate Functions — สรุปข้อมูล

Function	ทำอะไร	ตัวอย่าง
`COUNT(*)`	นับแถว	`SELECT COUNT(*) FROM students`
`COUNT(col)`	นับ non-NULL ใน column	`COUNT(email)` = คนที่มี email
`SUM(col)`	รวม	`SUM(score)`
`AVG(col)`	เฉลี่ย	`AVG(gpa)`
`MIN(col)` / `MAX(col)`	น้อยสุด / มากสุด	`MAX(score)`
`GROUP BY`	จัดกลุ่มก่อน aggregate	`GROUP BY major`
`HAVING`	filter หลัง GROUP BY (WHERE filter ก่อน)	`HAVING COUNT(*) > 5`

import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() cur.executescript(""" CREATE TABLE scores ( student_id TEXT, subject TEXT, score INTEGER, major TEXT ); INSERT INTO scores VALUES ('U1', 'Drawing', 85, 'MAE'), ('U2', 'Drawing', 70, 'EE'), ('U3', 'Drawing', 92, 'MAE'), ('U4', 'Drawing', 60, 'ChE'), ('U1', 'Lab', 78, 'MAE'), ('U2', 'Lab', 82, 'EE'), ('U3', 'Lab', 88, 'MAE'); """) print("=== ทั้งหมด ===") print(list(cur.execute("SELECT COUNT(*), AVG(score), MAX(score), MIN(score) FROM scores"))) print("\n=== กลุ่มตามวิชา (GROUP BY) ===") for row in cur.execute(""" SELECT subject, COUNT(*) AS n, ROUND(AVG(score), 1) AS avg, MAX(score) AS hi FROM scores GROUP BY subject """): print(row) print("\n=== กลุ่มตามสาขา · เฉพาะที่มี > 1 คน (HAVING) ===") for row in cur.execute(""" SELECT major, COUNT(*) AS n, ROUND(AVG(score), 1) AS avg FROM scores GROUP BY major HAVING COUNT(*) > 1 ORDER BY avg DESC """): print(row)

🕳 3) NULL Handling + CASE/WHEN

Real-world data มี NULL เสมอ · SQL จัดการ NULL พิเศษ (NULL ≠ 0 · NULL ≠ "")

import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() cur.executescript(""" CREATE TABLE students ( id TEXT PRIMARY KEY, name TEXT, email TEXT, -- บางคนยังไม่มี gpa REAL -- ปี 1 ยังไม่มี ); INSERT INTO students VALUES ('U1', 'Ploy', 'ploy@ubu.ac.th', 3.45), ('U2', 'Bank', NULL, 3.20), ('U3', 'Mint', 'mint@ubu.ac.th', NULL), ('U4', 'Tum', NULL, NULL); """) # IS NULL / IS NOT NULL — ห้ามใช้ = NULL print("=== คนที่ไม่มี email ===") print(list(cur.execute("SELECT name FROM students WHERE email IS NULL"))) print("\n=== คนที่มีทั้ง email และ gpa ===") print(list(cur.execute("SELECT name FROM students WHERE email IS NOT NULL AND gpa IS NOT NULL"))) # COALESCE — แทน NULL ด้วยค่า default print("\n=== COALESCE — แสดง 'N/A' แทน NULL ===") for row in cur.execute(""" SELECT name, COALESCE(email, 'N/A') AS email, COALESCE(gpa, 0.0) AS gpa FROM students """): print(row) # CASE/WHEN — if-else ใน SQL print("\n=== CASE/WHEN — แปลง gpa เป็น grade band ===") for row in cur.execute(""" SELECT name, gpa, CASE WHEN gpa IS NULL THEN 'ยังไม่มี' WHEN gpa >= 3.5 THEN 'ดีเยี่ยม' WHEN gpa >= 3.0 THEN 'ดี' WHEN gpa >= 2.0 THEN 'พอใช้' ELSE 'อ่อน' END AS band FROM students """): print(row)

⚠️ WHERE x = NULL ไม่ทำงาน! NULL เปรียบเทียบเท่าไม่ได้ · ต้องใช้ IS NULL / IS NOT NULL · bug นี้พบบ่อยมาก · AI ก็เขียนผิดบางครั้ง

🔄 4) Transactions — All-or-Nothing

ถ้าโปรแกรม crash กลาง INSERT 100 แถว — อะไรเกิดขึ้น? · ครึ่งหนึ่งใน · ครึ่งหนึ่งหาย · transaction ป้องกัน

import sqlite3 conn = sqlite3.connect(":memory:") cur = conn.cursor() cur.execute("CREATE TABLE accounts (id TEXT, balance INTEGER)") cur.execute("INSERT INTO accounts VALUES ('A', 1000), ('B', 500)") conn.commit() # ตัวอย่าง — โอนเงิน A → B def transfer(amount): try: cur.execute("BEGIN") cur.execute("UPDATE accounts SET balance = balance - ? WHERE id = 'A'", (amount,)) if amount > 800: raise ValueError("จำลอง error กลาง transaction!") cur.execute("UPDATE accounts SET balance = balance + ? WHERE id = 'B'", (amount,)) conn.commit() # ✅ ทุกอย่างผ่าน → save print(f"✅ โอน {amount} สำเร็จ") except Exception as e: conn.rollback() # ❌ rollback ทุกการเปลี่ยน print(f"❌ {e} → rollback · เงินกลับเดิม") print("ก่อน:") print(list(cur.execute("SELECT * FROM accounts"))) transfer(100) # ✅ ผ่าน print("\nหลังโอน 100:") print(list(cur.execute("SELECT * FROM accounts"))) transfer(900) # ❌ error → rollback print("\nหลังพยายามโอน 900 (พังกลางทาง):") print(list(cur.execute("SELECT * FROM accounts"))) # A ยังคงมี 900 — ไม่ใช่ -800!

ACID — 4 คุณสมบัติของ Database

Atomic — transaction = all-or-nothing
Consistent — data จะ valid ตามกฎเสมอ (constraints)
Isolated — 2 transaction รันพร้อมกันไม่ปนกัน
Durable — commit แล้ว = save แน่นอน · แม้ไฟดับ

SQLite, PostgreSQL, MySQL = ACID · NoSQL ส่วนใหญ่ ไม่ ACID เต็มที่ (eventual consistency)

📇 5) Indexes — เร็วขึ้น 100 เท่า เมื่อข้อมูลเยอะ

ตาราง 10 ล้านแถว · WHERE email = 'x' = scan ทุกแถว = ช้า · สร้าง index = เหมือนสารบัญหนังสือ = เร็ว

-- สร้าง index บน column ที่ query บ่อย
CREATE INDEX idx_students_email ON students(email);
CREATE INDEX idx_scores_student_id ON scores(student_id);

-- composite index (หลาย column)
CREATE INDEX idx_scores_subj_score ON scores(subject, score);

-- ดู query plan
EXPLAIN QUERY PLAN
SELECT * FROM students WHERE email = 'ploy@ubu.ac.th';
-- → "USING INDEX idx_students_email" = เร็ว ✅

⚠️ Index ไม่ฟรี Index = ใช้ disk space · INSERT/UPDATE ช้าขึ้น (ต้อง update index ด้วย) · กฎ: สร้าง index บน column ที่ query บ่อย เท่านั้น · ไม่ใช่ทุก column · Year 1 ไม่ต้องลึก · "รู้ว่ามี" + "เมื่อช้าให้คิดถึง"

⚛️ ORM — เขียน Python · ได้ SQL

ORM (Object-Relational Mapping) = library ที่ "แปลง Python class ↔ SQL table" · ไม่ต้องเขียน SQL ดิบ · ทุก framework เว็บ (Django, FastAPI, Flask-SQLAlchemy) ใช้ ORM

📊 SQL ดิบ vs ORM

# SQL ดิบ — เขียน string import sqlite3 conn = sqlite3.connect("app.db") cur = conn.cursor() cur.execute(""" SELECT s.name, AVG(sc.score) FROM students s JOIN scores sc ON s.id = sc.student_id WHERE s.major = ? GROUP BY s.id HAVING AVG(sc.score) > 75 """, ("MAE",)) results = cur.fetchall() for name, avg in results: print(name, avg)

# SQLAlchemy ORM — เขียน Python from sqlalchemy import create_engine, func from sqlalchemy.orm import Session from models import Student, Score # class ของเรา engine = create_engine("sqlite:///app.db") with Session(engine) as session: results = ( session.query(Student.name, func.avg(Score.score)) .join(Score) .filter(Student.major == "MAE") .group_by(Student.id) .having(func.avg(Score.score) > 75) .all() ) for name, avg in results: print(name, avg)

🎯 เมื่อไหร่ใช้ ORM vs SQL ดิบ

สถานการณ์	ใช้
app เล็ก · query ง่าย · 1 ตาราง	SQL ดิบ · ใช้ `sqlite3` + `pandas`
app กลาง · หลาย table · CRUD เยอะ	ORM (SQLAlchemy, Tortoise, Django ORM)
query ซับซ้อนมาก · subquery · window function	SQL ดิบ · เพราะ ORM แปลออกมาแย่
ทีมใหญ่ · ต้อง migration	ORM + Alembic
เรียนรู้ครั้งแรก	SQL ดิบก่อน · เข้าใจดีกว่า · ค่อยขึ้น ORM

🗂 Database Migrations (Alembic) — รู้ว่ามี

เมื่อ schema เปลี่ยน (เพิ่ม column · rename · drop) ใน production — migration = script อัพเดต database ทีละ version · เก็บใน git · rollback ได้ · ใช้ Alembic ที่จับคู่กับ SQLAlchemy

alembic init migrations
alembic revision --autogenerate -m "add email to students"
alembic upgrade head        # apply
alembic downgrade -1        # rollback

🌸 NoSQL — เมื่อ SQL ไม่เหมาะ

NoSQL = database ที่ไม่ใช่ตารางแบบ relational · มีหลายแบบ · เลือกใช้เมื่อ "data structure ไม่เหมาะกับตาราง"

NoSQL	เก็บแบบ	เหมาะกับ	ตัวอย่าง
MongoDB	Document (JSON)	schema ไม่นิ่ง · nested data · social feed · CMS	Atlas free tier · `pymongo`
Redis	Key-Value (in-memory)	cache · session · rate limit · queue · pub/sub	Upstash free · `redis-py`
Firestore	Document + real-time	mobile app · real-time chat · IoT live data	Firebase free tier
InfluxDB	Time-series	sensor data · IoT logs · metrics	InfluxDB Cloud free
Vector DB (Pinecone, Qdrant)	Embeddings	RAG · semantic search · AI memory	ตัวเลือกใหม่ในยุค AI

📊 SQL vs NoSQL — เมื่อไหร่เลือก

ดี ๆ กับ	SQL (Postgres/SQLite)	NoSQL (Mongo/Redis/Firestore)
Schema นิ่ง · structured	✅	❌
Schema เปลี่ยนบ่อย · flexible	❌	✅
JOIN หลายตาราง	✅	⚠️ ลำบาก
ACID transactions	✅	⚠️ ส่วนใหญ่ eventual
Real-time updates	⚠️	✅ (Firestore)
Cache / session	❌	✅ Redis
Time-series sensor data	⚠️	✅ InfluxDB
AI embeddings · semantic search	⚠️ pgvector	✅ Vector DB

📌 สำหรับ Final Project เริ่มที่ SQLite ก่อนเสมอ · ถ้ามี IoT/sensor + real-time → ลอง Firestore · ถ้าทำ AI chatbot ที่ต้อง "จำเรื่องที่คุยไปแล้ว" → ลอง vector DB (W12 LLM section) · อย่ารีบไป NoSQL ถ้า SQL พอ

🧪 Workshop — Grading App End-to-End Data Pipeline

ใน 1 notebook — เริ่มจาก list of dict → แปลง JSON → CSV → SQLite → query กลับ · ใช้ข้อมูล Grading App ของพี่นัทเป็น "ของจริง"

import pandas as pd import json, csv, sqlite3 # ==== STEP 1: สร้างข้อมูลจาก list of dicts ==== students = [ {"id": "640001", "name": "Ploy", "year": 2, "major": "MAE", "gpa": 3.45}, {"id": "640002", "name": "Bank", "year": 2, "major": "EE", "gpa": 3.20}, {"id": "640003", "name": "Mint", "year": 2, "major": "MAE", "gpa": 3.78}, {"id": "640004", "name": "Tum", "year": 1, "major": "ChE", "gpa": 2.90}, {"id": "640005", "name": "Aof", "year": 3, "major": "MAE", "gpa": 3.60}, ] print(f"✅ Step 1: สร้าง {len(students)} students ใน RAM") # ==== STEP 2: เซฟลง JSON ==== with open("students.json", "w", encoding="utf-8") as f: json.dump(students, f, ensure_ascii=False, indent=2) print("✅ Step 2: เขียน students.json") # ==== STEP 3: แปลงเป็น CSV ผ่าน Pandas ==== df = pd.DataFrame(students) df.to_csv("students.csv", index=False, encoding="utf-8") print("✅ Step 3: แปลงเป็น students.csv") print(df) # ==== STEP 4: ใส่ลง SQLite ==== conn = sqlite3.connect("grading.db") df.to_sql("students", conn, if_exists="replace", index=False) print("\n✅ Step 4: ใส่ลง SQLite (table 'students')") # ==== STEP 5: ตอบ 3 คำถามด้วย SQL ==== print("\n--- Q1: นักศึกษาในแต่ละสาขา ---") print(pd.read_sql("SELECT major, COUNT(*) as cnt FROM students GROUP BY major", conn)) print("\n--- Q2: gpa เฉลี่ยของแต่ละชั้นปี ---") print(pd.read_sql("SELECT year, AVG(gpa) as avg_gpa FROM students GROUP BY year", conn)) print("\n--- Q3: top 3 GPA ---") print(pd.read_sql("SELECT name, gpa FROM students ORDER BY gpa DESC LIMIT 3", conn)) conn.close() print("\n🎉 ครบ pipeline · ลองแก้ข้อมูล หรือคำถาม แล้ว Run ใหม่")

📝 ส่วนต่อยอด (ใน Cursor — งานบ้าน)

copy code นี้ลง Jupyter notebook (.ipynb) — แยกเป็น 5 cell ตาม step
หา data จริงของคุณ — Google Forms ที่เคยทำ · sensor log · expense tracking · datase ที่ data.go.th
ออกแบบ schema ตาม 4 คำถามด้านบน · วาด ER diagram ใน Mermaid
แทนข้อมูล Grading App ด้วยของคุณ — รันทั้ง pipeline ใหม่
ตอบ 3 คำถามใหม่ด้วย groupby / merge / value_counts
commit + push GitHub — README.md อธิบาย data source + schema + คำถาม

📌 ตัดสินใจ — ใช้อันไหนเมื่อไหร่

สถานการณ์	ใช้	เหตุผล
ตัวแปรในโปรแกรม < 1000 records	list of dict	เร็ว · simple
เซฟ config / state ของ CLI tool	JSON	มนุษย์อ่านได้ · มี structure
ส่งออกให้ Excel	CSV หรือ .xlsx	มาตรฐาน · เปิดที่ไหนก็ได้
analyze ข้อมูล + plot	pandas DataFrame	function สำหรับ data ครบ
app ที่ต้อง search/filter ซับซ้อน	SQLite	SQL เร็ว · index
หลาย user · production	PostgreSQL (W12)	concurrent · network · backup

ข้อผิดที่พบบ่อย

กระโดดไป SQL ตั้งแต่แรก — โปรเจกต์ส่วนใหญ่ของนักศึกษาปี 1 ใช้ JSON ก็เพียงพอ · เริ่ม simple ก่อน

เซฟ JSON โดยไม่มี encoding="utf-8" ภาษาไทยพังทันที · กฎทอง: UTF-8 ทุกไฟล์

ลืม index=False เวลา to_csv Pandas จะใส่ column ใหม่ "Unnamed: 0" · กฎ: df.to_csv(..., index=False) เสมอ

ส่งงานสัปดาห์นี้

📓 Jupyter/Colab notebook ที่ทำ Workshop ครบ 7 ขั้น
📂 ใน repo มี: ไฟล์ต้นฉบับ CSV + JSON ที่แปลงแล้ว + .db ที่ insert แล้ว
📝 README อธิบายว่าข้อมูลมาจากไหน + 3 คำถามที่ตอบ

Reference จาก slide เดิม

ครอบคลุม Topic เดิมที่กระจายอยู่ — list/dict (จาก Topic 7), CSV จาก Topic 10 (Pandas), และขยายไป JSON + SQLite ที่ slide เดิมไม่ครอบคลุม