Pandas & NumPy Academy · 강의

DataFrame을 데이터베이스 테이블에 쓰기

정제된 DataFrame을 DataFrame.to_sql()을 사용해 새 테이블이나 기존 테이블에 저장하고, if_exists와 청크 크기를 제어합니다.

DataFrame을 데이터베이스 테이블에 쓰기은(는) CoddyKit의 무료 Pandas & NumPy Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 Pandas & NumPy Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. Pandas & NumPy Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

Why Write DataFrames to Databases?

After cleaning and transforming data in Pandas, you often need to persist the results back to a database: to make them available to other applications, dashboards, or team members; to store incremental analysis results; or to build a data mart from a raw data lake. DataFrame.to_sql() is the standard Pandas method for writing data to any SQLAlchemy-supported database in a single call.

Basic to_sql() Usage

df.to_sql('table_name', con=engine, if_exists='replace', index=False) writes the DataFrame to a database table. The if_exists parameter controls what happens if the table already exists: 'fail' raises an error, 'replace' drops and recreates the table, and 'append' adds new rows without touching existing ones. Always set index=False unless you explicitly want to store the DataFrame index as a column in the database.

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')

The if_exists Parameter Explained

The three values of if_exists serve different use cases. 'replace' is for development: drop the old table and create a fresh one — schema changes are automatic but all old data is lost. 'append' is for incremental loads: add new rows to the existing table without changing its structure — useful for daily batch jobs. 'fail' is a safety guard: use it to protect important tables from being accidentally overwritten by a pipeline with a bug.

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')

Controlling Column Data Types

By default, to_sql() maps Pandas dtypes to SQLAlchemy types automatically. Sometimes the defaults are wrong — for example, a datetime64 column might be stored as TEXT in SQLite. Use the dtype parameter to specify exact SQL types using SQLAlchemy type objects. This ensures correct storage, proper indexing, and accurate type handling when the data is read back. Always verify the schema after writing with a quick PRAGMA table_info() or 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()})

Writing in Chunks with chunksize

For large DataFrames, to_sql() without a chunksize tries to insert all rows in a single statement, which can fail with a database timeout or memory error. Specify chunksize=N to insert N rows per transaction. This gives the database a chance to commit incrementally and reduces peak memory usage. A chunksize of 10,000–50,000 rows typically balances insert speed and memory, but the optimal value depends on your database and network latency.

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: Insert or Update

Pandas' to_sql() does not natively support upsert (insert if new, update if exists). To implement upsert, use SQLAlchemy's Core with an INSERT OR REPLACE (SQLite) or ON CONFLICT DO UPDATE (PostgreSQL) statement. The common workaround in Pandas is: write to a temporary staging table with if_exists='replace', then run raw SQL to merge the staging table into the production table, then drop the staging table.

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')

Verifying the Write

After writing, always verify the result by reading back a summary count and row count. Compare them against the source DataFrame. This catches silent failures caused by dtype mismatches (e.g., NaN in an integer column causing partial inserts) or database constraints (e.g., unique key violations silently skipping rows in some configurations). A quick SELECT COUNT(*) FROM table after every to_sql call adds minimal overhead and prevents silent data loss.

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!'

Adding a Primary Key After Writing

to_sql() writes data but does not add primary keys or database constraints — it creates a plain table. For a production table, add the primary key constraint separately using raw SQL executed through SQLAlchemy. SQLite requires recreating the table to add constraints after creation, but PostgreSQL supports ALTER TABLE ADD PRIMARY KEY. Alternatively, define the full schema upfront and use if_exists='append' to insert data into an existing properly-defined table.

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)

Transactional Writes

For data consistency, wrap to_sql() in an explicit transaction. If any step in a multi-table write fails, you can roll back all changes. Without a transaction, partial writes can leave the database in an inconsistent state. SQLAlchemy's connection context manager with conn.begin() enables manual transaction control. Alternatively, use engine.begin() for an auto-commit block that rolls back on exception.

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}')

Performance: Bulk Insert Methods

The default to_sql() inserts one row per SQL statement, which is very slow for large DataFrames. Pass method='multi' to use a single INSERT with multiple value tuples — typically 10-100x faster. For PostgreSQL, pass a custom method function that uses the COPY protocol (via psycopg2's copy_expert) for the absolute fastest bulk load. The optimal method depends on your database version and network setup.

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')

Logging and Auditing Writes

In production pipelines, track what was written and when by maintaining an audit log table. After each successful to_sql(), insert a row into the audit log with the table name, row count, timestamp, and pipeline run ID. This makes it easy to detect missing runs, double-writes, or schema changes over time. The audit log itself is a Pandas DataFrame written via to_sql — the same technique applied recursively for operational monitoring.

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')

Quick Check

Test your understanding of Data Analysis concepts from this lesson.

Lesson Recap

In this lesson you learned: df.to_sql() writes a DataFrame to any SQLAlchemy-connected database table with the if_exists parameter controlling create/append/replace behaviour, chunksize and method='multi' improve performance for large DataFrames, and transactional writes with engine.begin() ensure atomic multi-table updates that roll back on failure. Next up we compare Pandas and SQL to understand when each tool is the better choice.

자주 묻는 질문

“DataFrame을 데이터베이스 테이블에 쓰기” 강의는 무료인가요?

네 — “DataFrame을 데이터베이스 테이블에 쓰기” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 Pandas & NumPy Academy 강의 전체를 잠금 해제할 수 있습니다. Pandas & NumPy Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

“DataFrame을 데이터베이스 테이블에 쓰기”에서 뭘 배우나요?

정제된 DataFrame을 DataFrame.to_sql()을 사용해 새 테이블이나 기존 테이블에 저장하고, if_exists와 청크 크기를 제어합니다. 브라우저에서 직접 실행하는 실습 코드로 Pandas & NumPy Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

Pandas & NumPy Academy을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 Pandas & NumPy Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 3번째 강의입니다.

“DataFrame을 데이터베이스 테이블에 쓰기” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 Pandas & NumPy Academy 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 Pandas & NumPy Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← Pandas & NumPy Academy(으)로 돌아가기