การเขียน DataFrames ลงในตารางฐานข้อมูล
บันทึก DataFrame ที่ทำความสะอาดแล้วลงในตารางใหม่หรือตารางเดิมด้วย DataFrame.to_sql() พร้อมควบคุมค่า if_exists และ chunksize
การเขียน DataFrames ลงในตารางฐานข้อมูล เป็นบทเรียน Pandas & NumPy Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน Pandas & NumPy Academy และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
เหตุใดจึงต้องเขียน DataFrame ลงฐานข้อมูล
หลังจากทำความสะอาดและแปลงข้อมูลใน Pandas แล้ว คุณมักต้อง บันทึกผลลัพธ์ให้คงอยู่ กลับลงในฐานข้อมูล เพื่อให้แอปพลิเคชันอื่น แดชบอร์ด หรือสมาชิกในทีมเข้าถึงได้ เพื่อจัดเก็บผลการวิเคราะห์แบบเพิ่มทีละส่วน หรือเพื่อสร้างคลังข้อมูลเฉพาะเรื่องจากทะเลสาบข้อมูลดิบ DataFrame.to_sql() เป็นเมธอดมาตรฐานของ Pandas สำหรับเขียนข้อมูลลงในฐานข้อมูลที่ SQLAlchemy รองรับได้ด้วยการเรียกใช้เพียงครั้งเดียว
การใช้ to_sql() พื้นฐาน
df.to_sql('table_name', con=engine, if_exists='replace', index=False) จะเขียน DataFrame ลงในตารางฐานข้อมูล พารามิเตอร์ if_exists ควบคุมสิ่งที่จะเกิดขึ้นเมื่อตารางมีอยู่แล้ว: 'fail' จะทำให้เกิดข้อผิดพลาด 'replace' จะลบแล้วสร้างตารางขึ้นใหม่ และ 'append' จะเพิ่มแถวใหม่โดยไม่แก้ไขแถวเดิม ควรกำหนด index=False เสมอ เว้นแต่คุณต้องการจัดเก็บดัชนีของ DataFrame เป็นคอลัมน์ในฐานข้อมูลโดยเฉพาะ
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({
'date': pd.date_range('2024-01-01', periods=5),
'revenue': [1200.0, 980.5, 1450.0, 760.3, 1100.0],
'region': ['North', 'South', 'East', 'West', 'North']
})
df.to_sql('daily_revenue', con=engine,
if_exists='replace', index=False)
print('Table written successfully')คำอธิบายพารามิเตอร์ if_exists
ค่าทั้งสามของ if_exists เหมาะกับกรณีใช้งานที่แตกต่างกัน 'replace' เหมาะสำหรับการพัฒนา: ลบตารางเดิมแล้วสร้างตารางใหม่ — การเปลี่ยนแปลงโครงสร้างจะเกิดขึ้นโดยอัตโนมัติ แต่ข้อมูลเก่าทั้งหมดจะสูญหาย 'append' เหมาะสำหรับการโหลดข้อมูลเพิ่มทีละส่วน: เพิ่มแถวใหม่ลงในตารางเดิมโดยไม่เปลี่ยนโครงสร้าง — มีประโยชน์สำหรับงานแบบกลุ่มรายวัน 'fail' เป็นกลไกป้องกัน: ใช้เพื่อป้องกันไม่ให้ pipeline ที่มีข้อผิดพลาดเขียนทับตารางสำคัญโดยไม่ตั้งใจ
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
new_batch = pd.DataFrame({
'date': ['2024-06-01', '2024-06-02'],
'revenue': [1500.0, 1300.0],
'region': ['North', 'East']
})
# Append new rows without losing existing data
new_batch.to_sql('daily_revenue', con=engine,
if_exists='append', index=False)
print('Appended new rows')การควบคุมชนิดข้อมูลของคอลัมน์
โดยค่าเริ่มต้น to_sql() จะจับคู่ชนิดข้อมูลของ Pandas กับชนิดข้อมูลของ SQLAlchemy โดยอัตโนมัติ บางครั้งค่าเริ่มต้นอาจไม่ถูกต้อง — ตัวอย่างเช่น คอลัมน์ datetime64 อาจถูกจัดเก็บเป็น TEXT ใน SQLite ให้ใช้พารามิเตอร์ dtype เพื่อระบุชนิด SQL ที่แน่นอนด้วยออบเจ็กต์ชนิดข้อมูลของ SQLAlchemy วิธีนี้ช่วยให้จัดเก็บข้อมูลได้ถูกต้อง สร้างดัชนีได้เหมาะสม และจัดการชนิดข้อมูลได้แม่นยำเมื่อนำข้อมูลกลับมาอ่าน ควรตรวจสอบโครงสร้างเสมอหลังการเขียนด้วย PRAGMA table_info() หรือ inspector.get_columns() อย่างรวดเร็ว
import pandas as pd
import sqlalchemy as sa
from sqlalchemy import types
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Carol'],
'score': [0.95, 0.87, 0.91],
'created_at': pd.to_datetime(['2024-01-01', '2024-01-02', '2024-01-03'])
})
df.to_sql('users', con=engine, if_exists='replace', index=False,
dtype={'id': types.Integer(),
'score': types.Float(),
'created_at': types.DateTime()})การเขียนข้อมูลเป็นส่วนด้วย chunksize
สำหรับ DataFrame ขนาดใหญ่ to_sql() ที่ไม่ใช้ chunksize จะพยายามแทรกข้อมูลทุกแถวด้วยคำสั่งเดียว ซึ่งอาจล้มเหลวเนื่องจากหมดเวลาการทำงานของฐานข้อมูลหรือข้อผิดพลาดเกี่ยวกับหน่วยความจำ ให้ระบุ chunksize=N เพื่อแทรกข้อมูลครั้งละ N แถวต่อธุรกรรม วิธีนี้เปิดโอกาสให้ฐานข้อมูล commit ข้อมูลเป็นระยะ และลดการใช้หน่วยความจำสูงสุด โดยทั่วไป chunksize ขนาด 10,000–50,000 แถวจะสร้างสมดุลระหว่างความเร็วในการแทรกข้อมูลกับการใช้หน่วยความจำได้ดี แต่ค่าที่เหมาะสมที่สุดขึ้นอยู่กับฐานข้อมูลและความหน่วงของเครือข่ายของคุณ
import pandas as pd
import sqlalchemy as sa
import numpy as np
engine = sa.create_engine('sqlite:///results.db')
# Large DataFrame
df = pd.DataFrame({
'id': range(500000),
'value': np.random.randn(500000)
})
# Insert in chunks of 10,000 rows at a time
df.to_sql('large_table', con=engine,
if_exists='replace',
index=False,
chunksize=10000)
print('Written 500,000 rows')Upsert: แทรกหรืออัปเดต
to_sql() ของ Pandas ไม่รองรับ upsert โดยตรง (แทรกเมื่อเป็นข้อมูลใหม่ และอัปเดตเมื่อมีข้อมูลอยู่แล้ว) หากต้องการทำ upsert ให้ใช้ Core ของ SQLAlchemy ร่วมกับคำสั่ง INSERT OR REPLACE (SQLite) หรือ ON CONFLICT DO UPDATE (PostgreSQL) วิธีแก้ปัญหาที่ใช้กันทั่วไปใน Pandas คือ เขียนข้อมูลลงตารางชั่วคราวสำหรับพักข้อมูลด้วย if_exists='replace' จากนั้นเรียกใช้ SQL โดยตรงเพื่อรวมตารางพักข้อมูลเข้ากับตารางใช้งานจริง แล้วลบตารางพักข้อมูล
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
new_data = pd.DataFrame({
'id': [1, 2, 5],
'value': [99.9, 88.8, 77.7]
})
# Write to staging table
new_data.to_sql('staging', con=engine,
if_exists='replace', index=False)
# Merge into production (SQLite syntax)
with engine.connect() as conn:
conn.execute(sa.text(
'INSERT OR REPLACE INTO production SELECT * FROM staging'
))
conn.commit()
print('Upsert complete')การตรวจสอบการเขียนข้อมูล
หลังจากเขียนข้อมูลแล้ว ควร ตรวจสอบผลลัพธ์ เสมอด้วยการอ่านค่าจำนวนรวมและจำนวนแถวกลับมา แล้วเปรียบเทียบกับ DataFrame ต้นทาง วิธีนี้ช่วยตรวจพบความล้มเหลวที่ไม่แสดงข้อผิดพลาด ซึ่งเกิดจากชนิดข้อมูลไม่ตรงกัน (เช่น NaN ในคอลัมน์จำนวนเต็มทำให้แทรกข้อมูลได้เพียงบางส่วน) หรือข้อจำกัดของฐานข้อมูล (เช่น การละเมิดคีย์ที่ไม่ซ้ำกันซึ่งอาจทำให้บางแถวถูกข้ามไปโดยไม่แสดงข้อผิดพลาดในบางการตั้งค่า) การเรียกใช้ SELECT COUNT(*) FROM table อย่างรวดเร็วหลังการเรียกใช้ to_sql แต่ละครั้งใช้ทรัพยากรเพิ่มเพียงเล็กน้อย และช่วยป้องกันข้อมูลสูญหายโดยไม่รู้ตัว
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df = pd.DataFrame({'id': range(1000), 'value': range(1000)})
df.to_sql('my_table', con=engine, if_exists='replace', index=False)
# Verify
with engine.connect() as conn:
count = conn.execute(sa.text('SELECT COUNT(*) FROM my_table')).scalar()
print(f'Source rows: {len(df)}, DB rows: {count}')
assert count == len(df), 'Row count mismatch!'การเพิ่มคีย์หลักหลังการเขียนข้อมูล
to_sql() จะเขียนข้อมูล แต่ไม่เพิ่ม คีย์หลัก หรือข้อจำกัดของฐานข้อมูล — โดยจะสร้างตารางธรรมดา สำหรับตารางใช้งานจริง ให้เพิ่มข้อจำกัดคีย์หลักแยกต่างหากด้วย SQL โดยตรงที่ดำเนินการผ่าน SQLAlchemy SQLite จำเป็นต้องสร้างตารางใหม่เพื่อเพิ่มข้อจำกัดหลังจากสร้างตารางแล้ว แต่ PostgreSQL รองรับ ALTER TABLE ADD PRIMARY KEY อีกทางเลือกหนึ่งคือกำหนดโครงสร้างทั้งหมดไว้ล่วงหน้า แล้วใช้ if_exists='append' เพื่อแทรกข้อมูลลงในตารางที่มีการกำหนดโครงสร้างไว้อย่างถูกต้องแล้ว
import pandas as pd
import sqlalchemy as sa
from sqlalchemy import Table, Column, Integer, Float, MetaData
engine = sa.create_engine('sqlite:///results.db')
meta = MetaData()
# Define table with primary key
my_table = Table('defined_table', meta,
Column('id', Integer, primary_key=True),
Column('value', Float)
)
meta.create_all(engine) # Create table with constraints
# Then insert data using append
df = pd.DataFrame({'id': range(5), 'value': [1.1, 2.2, 3.3, 4.4, 5.5]})
df.to_sql('defined_table', con=engine,
if_exists='append', index=False)การเขียนข้อมูลแบบมีธุรกรรม
เพื่อให้ข้อมูลสอดคล้องกัน ให้ครอบ to_sql() ไว้ใน ธุรกรรม ที่ระบุไว้อย่างชัดเจน หากขั้นตอนใด ๆ ในการเขียนข้อมูลหลายตารางล้มเหลว คุณจะสามารถย้อนกลับการเปลี่ยนแปลงทั้งหมดได้ หากไม่มีธุรกรรม การเขียนข้อมูลเพียงบางส่วนอาจทำให้ฐานข้อมูลอยู่ในสถานะไม่สอดคล้องกัน ตัวจัดการบริบทการเชื่อมต่อของ SQLAlchemy ที่ใช้ร่วมกับ conn.begin() ช่วยให้ควบคุมธุรกรรมด้วยตนเองได้ อีกทางเลือกหนึ่งคือใช้ engine.begin() สำหรับบล็อกที่ commit โดยอัตโนมัติและย้อนกลับเมื่อเกิดข้อยกเว้น
import pandas as pd
import sqlalchemy as sa
engine = sa.create_engine('sqlite:///results.db')
df_orders = pd.DataFrame({'id': [1, 2], 'amount': [100.0, 200.0]})
df_summary = pd.DataFrame({'total': [300.0], 'count': [2]})
try:
with engine.begin() as conn: # Auto-rollback on exception
df_orders.to_sql('orders_v2', con=conn,
if_exists='replace', index=False)
df_summary.to_sql('summary_v2', con=conn,
if_exists='replace', index=False)
print('Both tables written atomically')
except Exception as e:
print(f'Write failed, rolled back: {e}')ประสิทธิภาพ: วิธีแทรกข้อมูลจำนวนมาก
โดยค่าเริ่มต้น to_sql() จะแทรกข้อมูลทีละแถวด้วยคำสั่ง SQL หนึ่งคำสั่งต่อแถว ซึ่งช้ามากสำหรับ DataFrame ขนาดใหญ่ ให้ส่ง method='multi' เพื่อใช้คำสั่ง INSERT เดียวที่มีชุดค่าหลายชุด — โดยทั่วไปจะเร็วขึ้น 10–100 เท่า สำหรับ PostgreSQL ให้ส่งฟังก์ชัน method แบบกำหนดเองที่ใช้โพรโทคอล COPY (ผ่าน copy_expert ของ psycopg2) เพื่อโหลดข้อมูลจำนวนมากให้เร็วที่สุด วิธีที่เหมาะสมที่สุดขึ้นอยู่กับรุ่นของฐานข้อมูลและการตั้งค่าเครือข่ายของคุณ
import pandas as pd
import sqlalchemy as sa
import numpy as np
import time
engine = sa.create_engine('sqlite:///perf.db')
df = pd.DataFrame({'a': range(100000), 'b': np.random.randn(100000)})
# Default (one row per INSERT) — slow
start = time.time()
df.to_sql('test_default', con=engine, if_exists='replace', index=False)
print(f'Default: {time.time()-start:.2f}s')
# multi-row INSERT — faster
start = time.time()
df.to_sql('test_multi', con=engine, if_exists='replace',
index=False, method='multi', chunksize=1000)
print(f'Multi: {time.time()-start:.2f}s')การบันทึกและตรวจสอบการเขียนข้อมูล
ใน pipeline สำหรับใช้งานจริง ให้ติดตามว่ามีการเขียนข้อมูลอะไรและเมื่อใด โดยดูแล ตารางบันทึกการตรวจสอบ หลังจาก to_sql() ทำงานสำเร็จแต่ละครั้ง ให้แทรกแถวหนึ่งแถวลงในบันทึกการตรวจสอบ โดยระบุชื่อตาราง จำนวนแถว เวลาประทับ และรหัสการทำงานของ pipeline วิธีนี้ช่วยให้ตรวจพบการทำงานที่หายไป การเขียนข้อมูลซ้ำ หรือการเปลี่ยนแปลงโครงสร้างเมื่อเวลาผ่านไปได้ง่าย ตารางบันทึกการตรวจสอบเองก็เป็น DataFrame ของ Pandas ที่เขียนผ่าน to_sql — เป็นการใช้เทคนิคเดิมซ้ำเพื่อเฝ้าติดตามการทำงานของระบบ
import pandas as pd
import sqlalchemy as sa
from datetime import datetime
engine = sa.create_engine('sqlite:///results.db')
def write_with_audit(df, table_name, engine, run_id):
df.to_sql(table_name, con=engine, if_exists='append', index=False)
audit = pd.DataFrame([{
'run_id': run_id,
'table_name': table_name,
'rows_written': len(df),
'written_at': datetime.utcnow().isoformat()
}])
audit.to_sql('audit_log', con=engine, if_exists='append', index=False)
print(f'Wrote {len(df)} rows to {table_name}')
df = pd.DataFrame({'id': [1, 2], 'val': [10, 20]})
write_with_audit(df, 'my_table', engine, run_id='run_001')ตรวจสอบความเข้าใจอย่างรวดเร็ว
ทดสอบความเข้าใจแนวคิดด้านการวิเคราะห์ข้อมูลจากบทเรียนนี้
สรุปบทเรียน
ในบทเรียนนี้ คุณได้เรียนรู้ว่า df.to_sql() เขียน DataFrame ลงในตารางฐานข้อมูลใด ๆ ที่เชื่อมต่อกับ SQLAlchemy โดยมีพารามิเตอร์ if_exists ควบคุมพฤติกรรมการสร้าง การเพิ่มข้อมูล และการแทนที่ข้อมูล, chunksize และ method='multi' ช่วยเพิ่มประสิทธิภาพสำหรับ DataFrame ขนาดใหญ่ และ การเขียนข้อมูลแบบมีธุรกรรมด้วย engine.begin() ช่วยให้การอัปเดตหลายตารางเป็นการดำเนินการแบบเป็นหนึ่งเดียวและย้อนกลับได้เมื่อล้มเหลว บทถัดไป เราจะเปรียบเทียบ Pandas กับ SQL เพื่อทำความเข้าใจว่าเมื่อใดควรเลือกใช้เครื่องมือแต่ละชนิด
คำถามที่พบบ่อย
บทเรียน “การเขียน DataFrames ลงในตารางฐานข้อมูล” ฟรีหรือไม่
ใช่ — ข้อความเต็มของ “การเขียน DataFrames ลงในตารางฐานข้อมูล” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส Pandas & NumPy Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
คุณจะเรียนรู้อะไรในบทเรียน “การเขียน DataFrames ลงในตารางฐานข้อมูล”
บันทึก DataFrame ที่ทำความสะอาดแล้วลงในตารางใหม่หรือตารางเดิมด้วย DataFrame.to_sql() พร้อมควบคุมค่า if_exists และ chunksize คุณปฏิบัติ Pandas & NumPy Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน
คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน Pandas & NumPy Academy หรือไม่
ไม่จำเป็นต้องมีประสบการณ์มาก่อน Pandas & NumPy Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน
บทเรียน “การเขียน DataFrames ลงในตารางฐานข้อมูล” ใช้เวลานานแค่ไหน
บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย
ฉันเขียนและรันโค้ดในบทเรียน Pandas & NumPy Academy นี้ได้ไหม
ได้ บทเรียน Pandas & NumPy Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ
บทเรียนทั้งหมดในหลักสูตรนี้
- การเชื่อมต่อฐานข้อมูลด้วย SQLAlchemy
- การเรียกใช้คำสั่ง SQL จาก Pandas
- การเขียน DataFrames ลงในตารางฐานข้อมูล
- Pandas เทียบกับ SQL: การเลือกเครื่องมือที่เหมาะสม