ข้ามไปยังเนื้อหา
Tayakorn

← คู่มือ Airflow — ทำไมงานหลายขั้นต้องมีคนคุมลำดับ

LEVEL 0 · ปูพื้นจากศูนย์

คืนที่สคริปต์พังตรงกลาง

ทุกต้นเดือน ร้านเครื่องเขียนสามสาขาส่งไฟล์ Excel นับสต็อกมาที่คุณ สาขาละไฟล์ แล้วสคริปต์ Python ตัวหนึ่งที่คุณเขียนไว้ก็ทำงานต่อจนจบ: รับไฟล์สามสาขา → รวมกันและตรวจว่าคอลัมน์ครบ → แปลงให้อยู่ในรูปเดียวกัน → โหลดเข้าฐานข้อมูล → ส่งรายงานยอดสต็อกรวม · Task Scheduler ปลุกมันตอนตีสองของวันที่ 1 ทุกเดือน และมันก็ทำแบบนั้นมาหลายเดือนโดยไม่มีใครต้องนึกถึงมัน

ร้านนี้ไม่มีอยู่จริง ชื่อสินค้าและตัวเลขทุกตัวในเล่มตั้งขึ้นใหม่ แต่รูปร่างของปัญหาทุกข้อที่จะเจอเป็นของจริง

1.1 เช้าวันที่ไฟล์ "ว่าง"

เช้าวันที่ 1 สิงหาคม รายงานยอดสต็อกเดือนกรกฎาคมถูกส่งออกไปแล้วตามเวลา สคริปต์จบโดยไม่มี error สักบรรทัด และรายงานบอกว่า สาขา ก เหลือของ 0 ชิ้น

คุณเปิดไฟล์ของสาขา ก ใน Excel แล้วเห็นแบบนี้

หน้าต่าง Excel เปิดไฟล์นับสต็อกสาขา ก เดือนกรกฎาคม เห็นแถว 1 ถึง 4 คือชื่อสาขา วันที่นับ 2026-07-31 แถวว่าง และหัวตาราง รหัส ชื่อสินค้า หมวด จำนวน จากนั้นเลขแถวกระโดดไปที่ 38 ปุ่มลูกศรที่หัวคอลัมน์จำนวนเป็นรูปกรวย และแถบสถานะด้านล่างขึ้นคำว่า Filter Mode

ชื่อสาขา วันที่นับ แล้วก็หัวตาราง ไม่มีสินค้าสักแถว ข้อสรุปที่ตามมาแทบจะอัตโนมัติคือสาขาส่งไฟล์เปล่ามา แล้วคุณก็พิมพ์ข้อความไปขอไฟล์ใหม่

ก่อนกดส่ง ลองดูเลขแถวทางซ้ายอีกครั้ง: 1 · 2 · 3 · 4 แล้วกระโดดไป 38 — แถว 5 ถึง 37 หายไปไหน · ปุ่มที่หัวคอลัมน์จำนวนไม่ใช่ลูกศรธรรมดาแต่เป็นรูปกรวย และแถบสถานะด้านล่างเขียนว่า Filter Mode

ไฟล์นี้ไม่ได้ว่าง มันถูกเซฟทั้งที่ตัวกรองยังค้างอยู่ คนที่สาขาเปิดตัวกรองหาสินค้าที่ของหมด (จำนวน = 0) แล้วเซฟปิดไปโดยไม่ได้ล้างตัวกรอง · เดือนนี้ไม่มีสินค้าไหนหมดเลยสักรายการ ทุกแถวจึงไม่ผ่านตัวกรองและถูกซ่อนหมด — ซ่อน ไม่ใช่ลบ ข้อมูลยังอยู่ในไฟล์ครบทุกแถว ขนาดไฟล์ก็ปกติ

⚠️ ความเข้าใจผิด: "เปิดแล้วเห็นแค่หัวตาราง แปลว่าไฟล์ว่าง" — อาการนี้เกิดขึ้นจริง คนเปิดไฟล์แล้วขอไฟล์ใหม่ทั้งที่ข้อมูลอยู่ครบ · ผิดเพราะสิ่งที่ Excel แสดงคือผลของตัวกรองที่บันทึกไว้ในไฟล์ ไม่ใช่เนื้อของไฟล์ · กลไกที่ถูกคือทุกแถวในไฟล์มีธง "ซ่อน" ของตัวเอง และตัวกรองที่ค้างอยู่ทำให้แถวที่ไม่เข้าเงื่อนไขติดธงนั้น · พิสูจน์ได้ด้วยการอ่านไฟล์ตรง ๆ ไม่ผ่านหน้าจอ แบบบล็อกข้างล่าง

บล็อกนี้สร้างไฟล์หน้าตาเดียวกับของสาขา ก (33 รายการ ตัวกรอง "จำนวน = 0" ค้างอยู่) แล้วเปิดกลับมาอ่านสองแบบ — อ่านทุกแถว กับอ่านเฉพาะแถวที่มองเห็น ซึ่งคือสิ่งที่ตาคุณเห็นบนจอ · ต้องมี openpyxl ก่อน (pip install openpyxl)

import tempfile
from pathlib import Path

from openpyxl import Workbook, load_workbook

# ── สร้างไฟล์หน้าตาเดียวกับของสาขา ก เดือนกรกฎาคม: 33 รายการ ตัวกรอง "จำนวน = 0" ค้างอยู่ ──
path = Path(tempfile.mkdtemp()) / "stock_2026-07_a.xlsx"
wb = Workbook()
ws = wb.active
ws.title = "สต็อก"
ws["A1"] = "นับสต็อก สาขา ก"
ws["A2"], ws["B2"] = "วันที่นับ", "2026-07-31"
for col, name in enumerate(["รหัส", "ชื่อสินค้า", "หมวด", "จำนวน"], start=1):
    ws.cell(4, col, name)  # แถว 4 = หัวตาราง
for i in range(1, 34):
    ws.append([f"S{i:03d}", f"สินค้า {i}", "เครื่องเขียน", 3 + (i % 3 != 0)])
ws.auto_filter.ref = "A4:D37"
ws.auto_filter.add_filter_column(3, ["0"])  # คนก่อนหน้ากรองหาของที่หมด แล้วเซฟทั้งที่ตัวกรองยังค้าง
for r in range(5, 38):
    ws.row_dimensions[r].hidden = True  # เดือนนี้ไม่มีของหมดสักรายการ ทุกแถวจึงไม่ผ่านตัวกรอง
wb.save(path)

# ── เปิดไฟล์กลับมาอ่านสองแบบ ──
ws = load_workbook(path).active
rows = range(5, ws.max_row + 1)
hidden = [r for r in rows if ws.row_dimensions[r].hidden]
visible = [r for r in rows if not ws.row_dimensions[r].hidden]
print(f"ตัวกรองค้างอยู่ที่ช่วง {ws.auto_filter.ref}")
print(f"แถวข้อมูลในไฟล์ {len(rows)} แถว · ติดธงซ่อน {len(hidden)} แถว")
print(f"อ่านทุกแถว → {len(rows)} แถว · {sum(ws.cell(r, 4).value for r in rows)} ชิ้น")
print(f"อ่านเฉพาะแถวที่มองเห็น → {len(visible)} แถว · {sum(ws.cell(r, 4).value for r in visible)} ชิ้น")

ผลรันจริง

ตัวกรองค้างอยู่ที่ช่วง A4:D37
แถวข้อมูลในไฟล์ 33 แถว · ติดธงซ่อน 33 แถว
อ่านทุกแถว → 33 แถว · 121 ชิ้น
อ่านเฉพาะแถวที่มองเห็น → 0 แถว · 0 ชิ้น

ข้อมูล 121 ชิ้นอยู่ในไฟล์ครบ แต่ตัวอ่านที่เคารพธงซ่อนได้ศูนย์

แล้วสคริปต์ของคุณได้ศูนย์มาจากไหน — สมมติว่าปีก่อนสาขาชอบซ่อนแถวสินค้าที่เลิกขายไว้ แล้วยอดรวมเกินจริง คุณเลยเติมบรรทัดเดียวให้สคริปต์ข้ามแถวที่ซ่อน บรรทัดนั้นถูกต้องสำหรับปัญหาเดือนนั้น และกลายเป็นตัวกลืนข้อมูลทั้งสาขาในเดือนนี้ · คนที่ไม่ได้ใช้สคริปต์ก็เจอแบบเดียวกัน: ลอง copy ตารางที่กรองค้างอยู่ใน Excel ไปวางที่อื่น จะได้หัวตารางมาแถวเดียว

เดือนนี้สคริปต์ไม่ได้พัง ไม่มี error · รหัสจบการทำงานเป็น 0 · รายงานออกตรงเวลา และผิด · ในระบบไม่มีอะไรรู้ว่ามันผิด จนกว่าคนอ่านรายงานจะทักมา

💡 ตัวอ่านควรทำยังไงกับแถวที่ซ่อน — ข้ามทิ้ง หรือนับรวม — ไม่มีคำตอบเดียว เพราะแถวซ่อนตัดได้สองทาง บทที่ 8 จะกลับมาที่ลำดับที่ถูก: อ่านทุกแถว ติดป้ายว่าแถวไหนซ่อน แล้วค่อยตัดสินทีละเรื่อง · ส่วนการกันไม่ให้ไฟล์ที่ได้ศูนย์แถวเดินผ่านไปถึงขั้นโหลด คือด่านตรวจของบทที่ 9

1.2 คืนที่มันพังจริง

เดือนกรกฎาคมพังแบบเงียบ แต่สคริปต์ตัวนี้เคยพังแบบเสียงดังมาก่อน และคืนนั้นสอนได้มากกว่า

ตีสองของวันที่ 1 มิถุนายน คืนปิดยอดเดือนพฤษภาคม ฐานข้อมูลปลายทางตัดการเชื่อมต่อระหว่างที่สคริปต์กำลังโหลดของสาขาที่สาม

บล็อกข้างล่างมีสองส่วน: ฟังก์ชัน monthly() คือสคริปต์ทั้งก้อน ทุกขั้นเรียงกันตามบรรทัดในฟังก์ชันเดียว · ส่วนล่างเรียกมันเป็นอีกโปรเซสแบบที่ Task Scheduler เรียก แล้วดูว่ามันทิ้งอะไรไว้ให้บ้าง — ครั้งแรกตอนฐานข้อมูลล่ม ครั้งที่สองตอนเช้าที่ฐานข้อมูลกลับมาแล้ว และคุณสั่งรันใหม่ทั้งก้อน

import sqlite3
import subprocess
import sys
import tempfile
from pathlib import Path

BRANCHES = {"ก": 120, "ข": 340, "ค": 560}  # ยอดที่แต่ละสาขานับได้เดือนพฤษภาคม (ชิ้น)


def monthly(work):
    """สคริปต์ปิดยอดรายเดือนแบบก้อนเดียว — ทุกขั้นอยู่ในฟังก์ชันนี้ เรียงกันตามบรรทัด"""
    files = {}
    for branch, total in BRANCHES.items():  # รับไฟล์ 3 สาขา (สาขาละ 33 รายการ)
        q, extra = divmod(total, 33)
        files[branch] = [q + (i < extra) for i in range(33)]
    rows = [(b, i + 1, n) for b, counts in files.items() for i, n in enumerate(counts)]  # รวม
    db = sqlite3.connect(work / "stock.db")
    db.execute("CREATE TABLE IF NOT EXISTS stock (branch, item, qty)")
    for branch in BRANCHES:  # โหลดทีละสาขา
        if branch == "ค" and not (work / "db-fixed").exists():
            raise ConnectionError("ฐานข้อมูลปลายทางตัดการเชื่อมต่อ")
        db.executemany("INSERT INTO stock VALUES (?, ?, ?)", [r for r in rows if r[0] == branch])
        db.commit()
    total = db.execute("SELECT SUM(qty) FROM stock").fetchone()[0]
    print(f"ส่งรายงาน: สต็อกรวม {total:,} ชิ้น")  # แจ้งผล


if len(sys.argv) == 2:  # ← ตัวลูก: คือสคริปต์ที่ Task Scheduler เรียกตอนตีสอง
    monthly(Path(sys.argv[1]))
    sys.exit()

# ── ตัวแม่: เรียกสคริปต์ข้างบนเป็นอีกโปรเซส แล้วดูว่าเหลืออะไรไว้ให้เราบ้าง ──
work = Path(tempfile.mkdtemp())


def run(label):
    r = subprocess.run([sys.executable, "-X", "utf8", __file__, str(work)],
                       capture_output=True, text=True, encoding="utf-8")
    print(f"── {label} ──")
    print(f"รหัสจบการทำงาน: {r.returncode}")
    print("ข้อความสุดท้าย:", (r.stdout or r.stderr).strip().splitlines()[-1])
    db = sqlite3.connect(work / "stock.db")
    for branch, n, qty in db.execute("SELECT branch, COUNT(*), SUM(qty) FROM stock GROUP BY branch ORDER BY branch"):
        print(f"  ในฐานข้อมูล สาขา {branch}: {n} แถว · {qty:,} ชิ้น")
    db.close()


run("คืนวันที่ 1 มิถุนายน 02:00")
(work / "db-fixed").touch()  # เช้ามาฐานข้อมูลกลับมาทำงานแล้ว
run("เช้าวันเดียวกัน — รันใหม่ทั้งก้อน")

ผลรันจริง

── คืนวันที่ 1 มิถุนายน 02:00 ──
รหัสจบการทำงาน: 1
ข้อความสุดท้าย: ConnectionError: ฐานข้อมูลปลายทางตัดการเชื่อมต่อ
  ในฐานข้อมูล สาขา ก: 33 แถว · 120 ชิ้น
  ในฐานข้อมูล สาขา ข: 33 แถว · 340 ชิ้น
── เช้าวันเดียวกัน — รันใหม่ทั้งก้อน ──
รหัสจบการทำงาน: 0
ข้อความสุดท้าย: ส่งรายงาน: สต็อกรวม 1,480 ชิ้น
  ในฐานข้อมูล สาขา ก: 66 แถว · 240 ชิ้น
  ในฐานข้อมูล สาขา ข: 66 แถว · 680 ชิ้น
  ในฐานข้อมูล สาขา ค: 33 แถว · 560 ชิ้น

ไล่ดูทีละบรรทัด

  • รหัสจบการทำงาน 1 กับบรรทัดสุดท้ายของ error คือทั้งหมดที่เหลือจากคืนนั้น — สิ่งที่ตัวตั้งเวลาอย่าง Task Scheduler หรือ cron ได้กลับมาจากโปรเซส มีแค่รหัสจบการทำงานกับข้อความที่มันพิมพ์ออกมา ไม่มีส่วนไหนบอกว่าข้างในทำไปถึงไหน
  • จะรู้ว่าโหลดไปแล้วสองสาขา ต้องเข้าไปเปิดฐานข้อมูลนับเอง — และต้องรู้ด้วยว่าควรนับอะไร
  • เช้ามารันใหม่ทั้งก้อน สคริปต์ทำทุกขั้นตั้งแต่ต้น รวมทั้งโหลดสาขา ก กับ ข ซ้ำ · รายงานบอก 1,480 ชิ้น ทั้งที่สามสาขานับรวมกันได้ 1,020 (120 + 340 + 560) · รอบนี้รหัสจบการทำงานเป็น 0 และรายงานผิด — อาการเดียวกับเดือนกรกฎาคม ต่างแค่ต้นเหตุ

💡 รันบล็อกในเล่มบน Windows แล้วส่งผลลงไฟล์ (python x.py > out.txt) ภาษาไทยกับเครื่องหมายบางตัวอาจทำให้ Python หยุดด้วย UnicodeEncodeError เพราะตอนส่งลงไฟล์ Python ใช้รหัสอักขระของระบบ ไม่ใช่ UTF-8 · ตั้งตัวแปร PYTHONUTF8=1 ก่อนรันแล้วหาย ส่วนการรันให้ผลขึ้นจอตรง ๆ ไม่มีปัญหานี้ · ส่วน -X utf8 ในบล็อกข้างบนคือสวิตช์เดียวกัน ใส่ให้โปรเซสลูกที่ถูกเรียกขึ้นมา

1.3 สามคำถามที่สคริปต์ตอบเองไม่ได้

สองเดือนนั้นทิ้งคำถามชุดเดียวกันไว้

คำถามเดือนพฤษภาคม (พังเสียงดัง)เดือนกรกฎาคม (พังเงียบ)
ทำถึงไหนแล้วอยู่ในตัวแปรของโปรเซสที่ตายไปแล้ว เหลือแค่ร่องรอยในฐานข้อมูลให้ไปนับเองสคริปต์คิดว่าทำครบทุกขั้น
ขั้นไหนรอขั้นไหนอยากรันแค่ "โหลดสาขา ค" แต่ข้อมูลที่แปลงแล้วอยู่ในหน่วยความจำ หายไปกับโปรเซส ต้องเริ่มใหม่ตั้งแต่รับไฟล์ขั้นโหลดเชื่อขั้นอ่านไฟล์ทุกอย่าง ไม่มีใครถามว่าศูนย์แถวสมเหตุสมผลไหม
ใครรู้ว่ามันพังรหัสจบการทำงาน 1 ในหน้าต่างที่ไม่มีใครเปิดดูตอนตีสองไม่มีใครเลย จนคนอ่านรายงานทัก

ทั้งสามคำถามมีคำตอบอยู่ครู่หนึ่ง — ขณะที่สคริปต์ยังรันอยู่ สคริปต์รู้ว่ากำลังทำขั้นไหน รู้ว่าขั้นก่อนหน้าเสร็จแล้ว รู้ว่าเพิ่งเจอ error · ปัญหาคือมันเก็บคำตอบทั้งหมดไว้ในตัวเอง: ลำดับขั้นอยู่ในลำดับบรรทัด · ความคืบหน้าอยู่ในตัวแปร · ความล้มเหลวอยู่ใน traceback ที่พิมพ์ออกจอ · พอโปรเซสจบหรือตาย คำตอบก็หายไปพร้อมกัน

ความรู้ว่างานทำถึงไหนแล้ว ต้องอยู่นอกตัวงาน — ไม่อย่างนั้นมันตายไปพร้อมกับงาน

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

1.4 ตัวคุมลำดับงานคืออะไร

💡 ตัวคุมลำดับงาน (workflow orchestrator) — โปรแกรมที่ถือคำตอบของสามคำถามนั้นไว้นอกตัวงาน: จดว่าแต่ละขั้นของแต่ละรอบอยู่ในสถานะไหน · รู้ว่าขั้นไหนต้องรอขั้นไหน · และเป็นคนสั่งรัน สั่งลองใหม่ จดบันทึก และบอกคนเมื่อมีอะไรพัง · Apache Airflow คือตัวหนึ่งในกลุ่มนี้ และเป็นตัวที่เล่มนี้ใช้ตั้งแต่บทที่ 5

สิ่งที่ตัวคุมทำไม่ได้สำคัญพอ ๆ กัน และบทนี้มีตัวอย่างให้ครบทั้งสองข้อแล้ว

  • มันไม่รู้ว่าข้อมูลผิด — เดือนกรกฎาคมทุกขั้นผ่านหมด ตัวคุมจะรายงานว่าสำเร็จเหมือนกัน · กันเรื่องนี้ต้องมีด่านตรวจก่อนโหลด ซึ่งคุณต้องเขียนเอง (บทที่ 9)
  • มันรันงานซ้ำแน่ — ลองใหม่ รันต่อจากจุดพัง ย้อนรันเดือนเก่า ล้วนเป็นการรันซ้ำ · ถ้าขั้นโหลดเขียนแบบต่อท้าย การรันซ้ำก็เบิ้ลแบบเช้าวันที่ 1 มิถุนายนทุกครั้ง · ทำให้แต่ละขั้นรันซ้ำแล้วได้ผลเท่าเดิมเป็นงานของคนเขียนขั้น ตัวคุมรับประกันให้ไม่ได้ (บทที่ 12)

1.5 คำถามที่บทนี้ทิ้งไว้

  1. รันใหม่ทั้งก้อนแล้วยอดเบิ้ล — ทำยังไงให้รันซ้ำแล้วได้ผลเท่าเดิม → บทที่ 12 ซึ่งเป็นบทพีคของเล่ม
  2. ถ้างานของคุณมีขั้นเดียว รันใหม่ทั้งก้อนได้โดยไม่เสียอะไร และไม่มีใครรอผล ยังต้องมีตัวคุมไหม → บทที่ 16
  3. ตัวอ่านควรทำยังไงกับแถวที่ซ่อน → บทที่ 8

ลองตอบโดยไม่ย้อนกลับไปอ่าน (เฉลยท้ายเล่ม)

  1. งานรายสัปดาห์ของคุณ: ดึงรายการสั่งซื้อจากระบบ → รวมเป็นไฟล์สรุป → ส่งอีเมลหาลูกค้า 200 คน · คืนหนึ่งมันพังตอนส่งไปได้ 80 ฉบับ ถ้าคุณรันใหม่ทั้งก้อน จะเกิดอะไรขึ้น และความรู้ข้อไหนที่หายไปกับโปรเซส
  2. ทำไมรหัสจบการทำงาน 0 ของเดือนกรกฎาคม ถึงไม่ได้แปลว่างานถูก
  3. เปิดไฟล์ Excel แล้วเห็นแค่หัวตาราง มีอะไรบนหน้าจอบ้างที่ต้องเช็กก่อนสรุปว่าไฟล์ว่าง

อ่านแบบเต็มเล่ม