Pandas & NumPy Academy · บทเรียน

การสตรีม CSV ด้วย chunksize

อ่าน CSV ขนาดใหญ่เป็นชิ้นขนาดคงที่ด้วย pd.read_csv(chunksize=) ประมวลผลแต่ละชิ้น และต่อหรือสะสมผลลัพธ์

การสตรีม CSV ด้วย chunksize เป็นบทเรียน Pandas & NumPy Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 1 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน Pandas & NumPy Academy และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน

ปัญหาของไฟล์ CSV ขนาดใหญ่

เมื่อไฟล์ CSV มีขนาดใหญ่กว่า RAM ที่มีอยู่ เช่น ไฟล์บันทึกขนาด 50 GB บนเครื่องที่มีหน่วยความจำ 16 GB การเรียกใช้ pd.read_csv('file.csv') จะล้มเหลวด้วย MemoryError หรือทำให้ระบบสลับข้อมูลไปยังพื้นที่จัดเก็บอย่างหนักจนทำงานช้ามากและแทบใช้งานไม่ได้ วิธีแก้คือ การอ่านเป็นส่วนย่อย แทนที่จะโหลดไฟล์ทั้งหมดในครั้งเดียว ให้ประมวลผลเป็นส่วนที่มีขนาดคงที่และสะสมผลลัพธ์ โดยไม่เก็บข้อมูลทั้งหมดไว้ในหน่วยความจำพร้อมกัน

พารามิเตอร์ chunksize ใน read_csv

การส่ง chunksize=N ให้กับ pd.read_csv() จะส่งคืนตัววนซ้ำ TextFileReader แทน DataFrame การวนซ้ำแต่ละครั้งจะให้ DataFrame ที่มีแถวไม่เกิน N แถว ไฟล์จะถูกอ่านแบบขี้เกียจ กล่าวคือ จะยังไม่มีการโหลดข้อมูลจนกว่าคุณจะขอส่วนถัดไป ตัววนซ้ำนี้ใช้ในลูป for หรือส่งต่อให้ pd.concat() ได้ ให้เลือก chunksize ที่ใหญ่พอสำหรับ I/O ที่มีประสิทธิภาพ เช่น 10,000–100,000 แถว แต่เล็กพอที่จะพอดีกับหน่วยความจำอย่างสบาย

import pandas as pd

# Returns a TextFileReader iterator, NOT a DataFrame
chunks = pd.read_csv('sales_data.csv', chunksize=10000)
print(type(chunks))  # <class 'pandas.io.parsers.readers.TextFileReader'>

for chunk in chunks:
    print(f'Chunk shape: {chunk.shape}')
    # process each chunk independently
    break   # just show the first chunk here

การประมวลผลแต่ละส่วนแยกจากกัน

รูปแบบที่พบบ่อยที่สุดคือดำเนินการ แปลงหรือกรองข้อมูล ในแต่ละส่วน เก็บผลลัพธ์ไว้ในรายการ แล้วจึงนำมาต่อกัน ตัวอย่างเช่น คุณอาจกรองแถวที่ตรงตามเงื่อนไข คำนวณสถิติของแต่ละส่วน หรือเลือกเฉพาะคอลัมน์ที่ต้องการ การทำงานกับชุดย่อยทำให้มีเพียงส่วนปัจจุบันเท่านั้นที่ใช้หน่วยความจำ ส่วนที่เหลือของไฟล์จะไม่ถูกแตะต้อง หลังจบลูป pd.concat(results) จะประกอบเป็น DataFrame สุดท้าย

import pandas as pd

results = []
for chunk in pd.read_csv('orders.csv', chunksize=50000):
    # Keep only high-value orders
    filtered = chunk[chunk['amount'] > 1000]
    results.append(filtered)

# Combine all filtered chunks
high_value = pd.concat(results, ignore_index=True)
print('High-value orders:', len(high_value))

การสะสมค่ารวมข้ามส่วน

บางครั้งคุณไม่จำเป็นต้องเก็บแถวใด ๆ ไว้เลย แต่ต้องการเพียง ค่ารวมสะสม ให้ติดตามผลรวม จำนวน หรือค่าต่ำสุด/สูงสุดสะสมข้ามส่วน โดยไม่สร้างรายการ DataFrame เพิ่มขึ้นเรื่อย ๆ รูปแบบนี้ใช้หน่วยความจำได้มีประสิทธิภาพที่สุด เพราะการใช้หน่วยความจำจะคงที่ไม่ว่าไฟล์จะมีขนาดเท่าใด เมื่อสิ้นสุด ให้คำนวณสถิติสุดท้ายจากตัวสะสมของคุณ

import pandas as pd

total_revenue = 0.0
total_rows = 0

for chunk in pd.read_csv('sales.csv', chunksize=100000):
    total_revenue += chunk['revenue'].sum()
    total_rows += len(chunk)

print(f'Processed {total_rows:,} rows')
print(f'Total revenue: ${total_revenue:,.2f}')

การระบุ dtype เพื่อเร่งการอ่านเป็นส่วน

โดยค่าเริ่มต้น Pandas จะอนุมานชนิดข้อมูลของคอลัมน์จากข้อมูล ซึ่งต้องสแกนแต่ละส่วนสองครั้ง ครั้งหนึ่งเพื่ออนุมาน และอีกครั้งเพื่อแยกวิเคราะห์ การระบุอาร์กิวเมนต์ dtype จะหลีกเลี่ยงค่าใช้จ่ายส่วนนี้ และยังป้องกันความไม่สอดคล้องของ dtype ระหว่างส่วนต่าง ๆ ด้วย ตัวอย่างเช่น คอลัมน์ที่มีจำนวนเต็มเป็นส่วนใหญ่แต่มีเซลล์ว่างหนึ่งเซลล์ อาจถูกอนุมานเป็น float64 ในส่วนหนึ่ง และเป็น object ในอีกส่วนหนึ่ง การระบุ dtype อย่างชัดเจนช่วยให้การอ่านทุกส่วนมีความสอดคล้องและรวดเร็วยิ่งขึ้น

import pandas as pd

dtype_map = {
    'order_id': 'int32',
    'customer_id': 'int32',
    'amount': 'float32',
    'category': 'category'
}

for chunk in pd.read_csv('orders.csv',
                         chunksize=50000,
                         dtype=dtype_map,
                         parse_dates=['order_date']):
    print(chunk.dtypes)
    break

การเลือกเฉพาะคอลัมน์ที่จำเป็น

ใช้พารามิเตอร์ usecols เพื่อโหลดเฉพาะคอลัมน์ที่การวิเคราะห์ของคุณต้องใช้ หาก CSV มี 50 คอลัมน์ แต่การหาค่ารวมใช้เพียง 3 คอลัมน์ ก็ไม่มีเหตุผลที่จะต้องแยกวิเคราะห์อีก 47 คอลัมน์ การใช้ usecols ร่วมกับ chunksize ช่วยลดทั้งเวลา I/O และการใช้หน่วยความจำได้อย่างมาก นี่เป็นหนึ่งในการปรับปรุงที่ทำได้ง่ายและให้ผลชัดเจนที่สุดสำหรับการประมวลผล CSV ขนาดใหญ่

import pandas as pd

# Only read the three columns we actually need
for chunk in pd.read_csv(
    'large_transactions.csv',
    chunksize=100000,
    usecols=['date', 'amount', 'region']
):
    print(chunk.columns.tolist())
    print(chunk.memory_usage(deep=True).sum() / 1e6, 'MB per chunk')
    break

การหาค่ารวมด้วย GroupBy ในแต่ละส่วน

การทำ การหาค่ารวมด้วย groupby ข้ามส่วนจำเป็นต้องสะสมผลลัพธ์บางส่วน ให้คำนวณ groupby ภายในแต่ละส่วน แล้วรวมผลลัพธ์ด้วย groupby ครั้งที่สองบนผลลัพธ์บางส่วนที่นำมาต่อกันแล้ว ตัวอย่างเช่น หากต้องการหายอดขายรวมตามภูมิภาคจากไฟล์ขนาด 10 GB ให้เก็บผลรวมตามภูมิภาคของแต่ละส่วนไว้ในรายการ จากนั้นนำมาต่อกันและจัดกลุ่มอีกครั้ง รูปแบบการหาค่ารวมสองรอบนี้บางครั้งเรียกว่า แนวทาง map-reduce

import pandas as pd

partials = []
for chunk in pd.read_csv('sales.csv',
                         chunksize=100000,
                         usecols=['region', 'revenue']):
    partial = chunk.groupby('region')['revenue'].sum()
    partials.append(partial)

# Combine partial sums
final = pd.concat(partials).groupby(level=0).sum()
print('Revenue by region:')
print(final.sort_values(ascending=False))

การจัดการข้อผิดพลาดในการแยกวิเคราะห์ข้ามส่วน

ไฟล์ CSV ขนาดใหญ่จากแหล่งภายนอกมักมี แถวที่มีรูปแบบไม่ถูกต้อง เช่น มีจุลภาคเกินมา การเข้ารหัสไม่ถูกต้อง หรือบรรทัดถูกตัด ใช้ on_bad_lines='skip' (Pandas 1.3 ขึ้นไป) หรือ error_bad_lines=False (Pandas รุ่นเก่า) เพื่อข้ามแถวที่มีปัญหาโดยไม่แสดงข้อความ และใช้ encoding='latin-1' หากการแยกวิเคราะห์ UTF-8 ล้มเหลว ติดตามว่าส่วนใดทำให้เกิดข้อผิดพลาดด้วย try-except รอบการประมวลผลของแต่ละส่วน เพื่อสร้างไปป์ไลน์ที่ทนทานและไม่หยุดทำงานเพราะแถวเสียเพียงแถวเดียวในไฟล์ที่มี 10 ล้านแถว

import pandas as pd

bad_chunks = []
all_chunks = []

for i, chunk in enumerate(pd.read_csv(
    'raw_data.csv',
    chunksize=50000,
    on_bad_lines='skip',
    encoding='utf-8',
    encoding_errors='replace'
)):
    try:
        # Your transformation here
        all_chunks.append(chunk)
    except Exception as e:
        bad_chunks.append((i, str(e)))
        print(f'Chunk {i} error: {e}')

print(f'Processed {len(all_chunks)} chunks, {len(bad_chunks)} errors')

การเขียนไฟล์ผลลัพธ์เป็นส่วนย่อย

เมื่อผลลัพธ์ที่ประมวลผลแล้วมีขนาดใหญ่เช่นกัน ให้เขียนผลลัพธ์แบบ ทยอยเพิ่ม แทนการสะสมทุกอย่างไว้ในหน่วยความจำแล้วเขียนในตอนท้าย เปิดไฟล์ CSV และเพิ่มแต่ละส่วนที่ประมวลผลแล้วด้วย mode='a' และใช้ header=False สำหรับส่วนถัดไป วิธีนี้ทำให้การใช้หน่วยความจำของไปป์ไลน์ผลลัพธ์คงที่ และทำให้คุณตรวจสอบผลลัพธ์บางส่วนได้ก่อนการทำงานทั้งหมดจะเสร็จสิ้น

import pandas as pd

first_chunk = True
for chunk in pd.read_csv('input.csv', chunksize=100000):
    # Transform
    processed = chunk[chunk['status'] == 'active'].copy()
    processed['revenue_usd'] = processed['revenue'] * 1.10

    # Write incrementally
    mode = 'w' if first_chunk else 'a'
    processed.to_csv('output.csv',
                     mode=mode,
                     header=first_chunk,
                     index=False)
    first_chunk = False

print('Done writing output.csv')

การประมาณขนาดส่วนที่เหมาะสมที่สุด

การเลือก chunksize เป็นการหาจุดสมดุล: หากเล็กเกินไปจะมีการวนลูป Python หลายครั้งและมีค่าใช้จ่ายสูง หากใหญ่เกินไปส่วนข้อมูลจะไม่พอดีกับ RAM แนวทางที่ใช้ได้จริงคือโหลดหนึ่งส่วน วัดการใช้หน่วยความจำด้วย chunk.memory_usage(deep=True).sum() และตั้งค่า chunksize ให้แต่ละส่วนใช้ RAM ที่มีอยู่ประมาณ 10–20% psutil.virtual_memory().available ของ Python จะแสดง RAM ที่พร้อมใช้งานขณะทำงาน ทำให้สามารถคำนวณ chunksize แบบปรับตามสภาพแวดล้อมได้

import pandas as pd

# Sample 1000 rows to estimate per-row memory
sample = pd.read_csv('big_file.csv', nrows=1000)
bytes_per_row = sample.memory_usage(deep=True).sum() / 1000
print(f'Bytes per row: {bytes_per_row:.0f}')

# Target: use at most 500 MB per chunk
target_bytes = 500 * 1024 * 1024
optimal_chunksize = int(target_bytes / bytes_per_row)
print(f'Recommended chunksize: {optimal_chunksize:,}')

การรวมผลลัพธ์จากส่วนต่าง ๆ อย่างมีประสิทธิภาพ

เมื่อสะสม DataFrame ของแต่ละส่วนไว้ในรายการแล้วนำมาต่อกัน โปรดทราบว่าการเรียกใช้ pd.concat กับเฟรมขนาดเล็กหลายร้อยเฟรมจะช้า เนื่องจากต้องจัดสรรหน่วยความจำซ้ำหลายครั้ง รูปแบบที่ดีกว่าคือ หาค่ารวมภายในแต่ละส่วน และจัดเก็บเฉพาะผลลัพธ์ที่รวมแล้วซึ่งมีขนาดเล็ก ไม่ใช่ส่วนข้อมูลทั้งหมด หากคุณจำเป็นต้องใช้ทุกแถวจริง ๆ การเขียนลงไฟล์ Parquet แบบทยอยเพิ่ม โดยใช้ pyarrow จะเร็วกว่าใช้ pd.concat ในตอนท้าย

import pandas as pd

# Efficient: aggregate first, small list of scalars
running_total = 0
running_count = 0

for chunk in pd.read_csv('sales.csv', chunksize=100000):
    running_total += chunk['amount'].sum()
    running_count += chunk['amount'].count()

print(f'Mean amount: {running_total / running_count:.2f}')

# Avoid: accumulating full chunk DataFrames
# results = []
# for chunk in reader:
#     results.append(chunk)   # memory grows to full file size
# df = pd.concat(results)     # slow for hundreds of chunks

ตรวจสอบความเข้าใจอย่างรวดเร็ว

ทดสอบความเข้าใจแนวคิดด้านการวิเคราะห์ข้อมูลจากบทเรียนนี้

สรุปบทเรียน

ในบทเรียนนี้ คุณได้เรียนรู้ว่า chunksize ใน pd.read_csv จะส่งคืนตัววนซ้ำของ DataFrame ซึ่งช่วยให้ประมวลผลไฟล์ขนาดใหญ่ได้โดยใช้หน่วยความจำอย่างมีประสิทธิภาพ อาร์กิวเมนต์ usecols และ dtype ช่วยลดการใช้หน่วยความจำต่อส่วนและเร่งการแยกวิเคราะห์ และ ตัวสะสมแบบต่อเนื่อง (sum, count, ค่าบางส่วน) ช่วยหลีกเลี่ยงการสร้างรายการที่เก็บทุกส่วน บทถัดไป เราจะศึกษาแนวทางการหาค่ารวมแบบทยอยเพิ่มข้ามส่วนต่าง ๆ อย่างละเอียดมากขึ้น

คำถามที่พบบ่อย

บทเรียน “การสตรีม CSV ด้วย chunksize” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การสตรีม CSV ด้วย chunksize” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส Pandas & NumPy Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การสตรีม CSV ด้วย chunksize”

อ่าน CSV ขนาดใหญ่เป็นชิ้นขนาดคงที่ด้วย pd.read_csv(chunksize=) ประมวลผลแต่ละชิ้น และต่อหรือสะสมผลลัพธ์ คุณปฏิบัติ Pandas & NumPy Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน Pandas & NumPy Academy หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน Pandas & NumPy Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 1 จากทั้งหมด 4 บทเรียน

บทเรียน “การสตรีม CSV ด้วย chunksize” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน Pandas & NumPy Academy นี้ได้ไหม

ได้ บทเรียน Pandas & NumPy Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

การสตรีม CSV ด้วย chunksize
การรวมค่าแบบเพิ่มทีละส่วนระหว่างชิ้นข้อมูล
บทนำสู่ Dask DataFrames
Parquet: การจัดเก็บข้อมูลแบบคอลัมน์ความเร็วสูง

← กลับไปที่ Pandas & NumPy Academy