Python สำหรับ Data Engineering

ทฤษฎีและกลไกการทำงาน (How it works): ในภาษา Python (CPython) วัตถุขนาดเล็กที่มีขนาดไม่เกิน 512 ไบต์ จะไม่ถูกร้องขอผ่านฟังก์ชัน malloc มาตรฐานของระบบปฏิบัติการโดยตรง เพื่อป้องกันปัญหาการหน่วงของสัญญาณระบบ (System Call Overhead) และการเกิดเศษซากรูในแรม (Memory Fragmentation) CPython จึงพัฒนาตัวจัดสรรหน่วยความจำชื่อ PyMalloc โครงสร้างการจัดเก็บของ PyMallocแบ่งออกเป็น 3 ชั้น: 1. **Block**: ขนาดคงที่ทวีคูณของ 8 ไบต์ (เช่น 8, 16, 24, จนถึง 512 ไบต์) สำหรับวางค่าวัตถุจริง 2. **Pool**: กลุ่มของบล็อกขนาดเดียวกัน มีขนาดรวมคงที่ 4KB ซึ่งตรงกับหน้าหน่วยความจำจริง (Virtual Memory Page Size) ของ OS 3. **Arena**: บล็อกหน่วยความจำหลักขนาด 256KB ที่จัดสรรตรงจากเคอร์เนลระบบปฏิบัติการ ประกอบด้วย Pool ย่อย ๆ ทั้งหมด 64 พูล

# Inspecting Python object memory allocation size in bytes
import sys

small_int = 42
large_list = [i for i in range(100000)]

print(f"Size of small integer: {sys.getsizeof(small_int)} bytes")
print(f"Size of large list object envelope: {sys.getsizeof(large_list)} bytes")
                    

Use Case ในชีวิตจริง (Real-world Scenario): เมื่อเราเขียนโปรแกรมดึงข้อมูลและทำความสะอาดไฟล์ JSON ขนาดใหญ่ที่มีวัตถุขนาดเล็กนับล้านตัว หากตัวแปรเหล่านั้นถูกสร้างและลบตลอดเวลา PyMalloc จะทำการ Recycle บล็อกหน่วยความจำภายใน Pool ทันทีโดยไม่ต้องรบกวน OS ทำให้โปรแกรมประมวลผลเร็วขึ้นและแรมไม่พุ่งสูง

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): ถึงแม้ PyMalloc จะช่วยกู้แรมระดับบล็อก แต่แรมชั้นบนสุดอย่าง **Arena** จะไม่ถูกคืนให้ระบบปฏิบัติการหลักจนกว่าพูลย่อยทั้งหมดภายในตัวมันจะว่างเปล่า 100% ส่งผลให้แรมของโปรเซสหลักดูค้างคาแม้ลบตัวแปรไปหมดแล้ว (Memory Bloat) แก้ไขโดยสลายแรมสะสมด้วยการรันบนโพรเซสย่อยแล้วปิดตัวลง หรือจัดพูลการใช้งานตัวแปรผ่าน `multiprocessing`

ทฤษฎีและกลไกการทำงาน (How it works): CPython มีกลไกเคลียร์ขยะสองขั้น: - **Reference Counting (การนับการอ้างอิง)**: ทุกอ็อบเจกต์มีตัวแปรร่วมภายใน `ob_refcnt` เมื่อเราชี้ตัวแปรใหม่ไปหาอ็อบเจกต์ ค่าจะเพิ่มขึ้น 1 และลดลง 1 เมื่อตัวแปรหลุดขอบเขตการทำงาน (Scope) เมื่อค่าเหลือ 0 แรมจะถูกทิ้งและปลดปล่อยทันที - **Generational Garbage Collection**: ใช้จัดการปัญหา **Cyclic References** (วัตถุอ้างอิงเป็นวงกลม เช่น A ชี้ B และ B ชี้ A ทำให้ค่า RC ค้างที่ 1 ไม่มีทางเป็น 0) ตัวเก็บขยะจะแบ่งอ็อบเจกต์เป็น 3 เจนเนอเรชัน (Gen 0, 1, 2) ข้อมูลใหม่จะอยู่ Gen 0 และถูกตรวจสอบบ่อยสุด หากรอดจากการล้างขยะจะถูกย้ายขึ้นชั้นถัดไปตามลำดับ

# Managing Garbage Collector manually for critical memory paths
import gc

# Disable automatic collection to prevent performance hiccups during tight loops
gc.disable()

# Perform memory intensive tasks...
# Manually trigger Gen 2 cleanup after processing is done
gc.collect(2)
gc.enable()
                    

Use Case ในชีวิตจริง (Real-world Scenario): ในงานเขียนคลาสจัดการการแปลงโมเดลข้อมูลขนาดใหญ่ที่ต้องเก็บสถานะความสัมพันธ์ต่อกัน (เช่น Parent-Child Nodes) ซึ่งมักเกิด Cyclic Reference การปิดระบบสแกนขยะอัตโนมัติในระหว่างลูปอ่านข้อมูลล้านบรรทัด และใช้วิธีสั่งรัน `gc.collect()` ด้วยมือตอนท้ายสุดจะทำให้สปีดการทำงานเรียบเนียนขึ้นถึง 20%

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การล้างขยะแบบอัตโนมัติอาจทำให้ระบบหน่วงกะทันหัน (Stop-the-world pauses) ในวินาทีที่ตัวสแกน Gen 2 เริ่มรัน แก้ไขโดยการแยกโครงสร้างตัวแปรไม่ให้ชี้วนซ้ำกัน หรือการประยุกต์ใช้ **Weak Reference** ผ่านไลบรารี `weakref` เพื่อให้อ้างอิงอ็อบเจกต์ได้โดยไม่เพิ่มคะแนน Reference Count

ทฤษฎีและกลไกการทำงาน (How it works): **GIL (Global Interpreter Lock)** คือตัวล็อกระดับล่ามภาษาใน CPython เพื่อป้องกันไม่ให้มี Thread มากกว่าหนึ่งตัวทำงานประมวลผลไบต์โค้ดพร้อมกัน เพื่อความปลอดภัยในการนับยอด Reference Counting สถาปัตยกรรมในการรันขนานจึงแบ่งออกเป็น: - **Multi-threading**: รันภายใต้โพรเซสเดียว เหมาะกับงาน **I/O-Bound** (เช่น โหลดหน้าเว็บ คิวรีข้อมูลระบบคลาวด์) เพราะระหว่างรอระบบส่งค่ากลับ เธรดจะคายสิทธิ์ล็อก GIL ให้เธรดอื่นรันแทน - **Multi-processing**: เลี่ยง GIL โดยการบิวด์ระบบโพรเซสของระบบปฏิบัติการขึ้นมาใหม่ มีพื้นที่แรมแยกและ CPython VM ของตนเอง รันบนซีพียูคนละคอร์ขนานกันได้อย่างสมบูรณ์แบบ (ขนานแท้จริง)

# Executing CPU-bound tasks in parallel using ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor
import math

def calculate_heavy_prime(n):
    # Simulating intensive computation
    count = 0
    for i in range(2, n):
        if all(i % d != 0 for d in range(2, int(math.sqrt(i)) + 1)):
            count += 1
    return count

if __name__ == "__main__":
    tasks = [100000, 100001, 100002, 100003]
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(calculate_heavy_prime, tasks))
    print(f"Results computed in parallel: {results}")
                    

Use Case ในชีวิตจริง (Real-world Scenario): การนำเข้าไฟล์พิกัดแผนที่พนักงานขับรถขนาด 10GB ที่ต้องถอดรหัสและคำนวณระยะพิกัดเชิงคณิตศาสตร์ (CPU-bound) ทีมงานใช้ `multiprocessing` แยกดึงข้อมูลกระจายรันขนานบนเซิร์ฟเวอร์ 16 คอร์ ทำให้ระยะเวลาการประมวลผลลดลงจาก 4 ชั่วโมง เหลือเพียง 20 นาที

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): การแชร์ข้อมูลระหว่างโพรเซสต้องผ่านการแปลงไบนารีวัตถุ (**Serialization / Pickling**) ซึ่งมีอัตราการดึงหน่วยความจำ I/O และซีพียูโชหุ้ยสูง (IPC Overhead) หากเราส่งตารางขนาดใหญ่ข้ามโพรเซส โปรแกรมอาจทำงานช้ากว่าโพรเซสเดียว แก้ไขโดยให้แต่ละโพรเซสย่อยอ่านเขียนไฟล์ปลายทางของตนเองโดยตรง หรือประยุกต์ใช้หน่วยความจำร่วมกัน (Shared Memory)

ทฤษฎีและกลไกการทำงาน (How it works): **Asyncio** นำเสนอการทำงานแบบร่วมมือพร้อมกันภายใต้เธรดเดียว (Cooperative Multitasking) โดยมีตัวควบคุมลูปเหตุการณ์ **Event Loop** คอยสลับงานฟังก์ชันที่รันค้างรอ I/O (เช่น รอข้อมูลอินเทอร์เน็ต) เมื่อฟังก์ชันพบคำสั่ง `await` ระบบจะคายสิทธิ์ให้ฟังก์ชันอื่นในคิวรันต่อทันทีโดยไม่ต้องรอระบบปฏิบัติการสลับบริบทเธรด (Context Switching) ซึ่งมีน้ำหนักเบาและกินแรมน้อยกว่าการใช้ Threading แบบปกติ

# Asynchronous parallel API ingestion client example using aiohttp
import asyncio
import aiohttp

async def fetch_api_page(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    urls = [f"https://api.pipecraft.dev/data?page={i}" for i in range(10)]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_api_page(session, url) for url in urls]
        # Execute all request pages concurrently
        results = await asyncio.gather(*tasks)
        print(f"Fetched {len(results)} API responses concurrently.")

if __name__ == "__main__":
    asyncio.run(main())
                    

Use Case ในชีวิตจริง (Real-world Scenario): ระบบดึงราคาสินทรัพย์การเงินเรียลไทม์จากปลายทางต่างประเทศ 50 แหล่งข้อมูลพร้อมกันเพื่อตรวจสอบความต่างของราคา (Arbitrage) การใช้ `asyncio` ร่วมกับ `aiohttp` ทำงานได้รวดเร็วเต็มประสิทธิภาพเน็ตเวิร์กโดยไม่เปลืองทรัพยากรเซิร์ฟเวอร์

ข้อควรระวังและวิธีแก้ (Pitfalls & Mitigations): หากมีโค้ดส่วนที่เป็นการคำนวณเชิงลึกแบบบล็อกการทำงาน (Blocking CPU-Bound Task) ไปรันอยู่ในลูปหลักของ Asyncio จะส่งผลให้ระบบงานทั้งหมดหยุดชะงัก (Event Loop Block) แก้ไขโดยโยนงานบล็อกนั้นไปรันข้างนอกผ่าน Thread/Process pool ด้วยคำสั่ง `loop.run_in_executor()`