SQL과 Python, 왜 함께 써야 할까?
データ 분석 실무에서 데이터는 대부분 관계형 데이터베이스(RDBMS)에 저장되어 있습니다. SQL로 데이터를 추출하고, Python으로 분석하는 워크플로우는 데이터 엔지니어와 분석가의 기본 무기입니다.
핵심 포인트: SQL은 데이터 추출에, Python은 변환·분석·시각화에 강합니다. 둘을 연결하는 다리가 바로 SQLAlchemy입니다.
이 글에서는 SQLAlchemy로 데이터베이스에 연결하고, Pandas로 분석한 뒤, 자동화 리포트까지 생성하는 end-to-end 워크플로우를 다룹니다.
Python-SQL 연동 도구 비교
어떤 라이브러리를 선택할지 먼저 정리해 보겠습니다.
| 라이브러리 | 특징 | ORM 지원 | Pandas 통합 | 추천 상황 |
|---|---|---|---|---|
| SQLAlchemy | 범용 ORM + Core | ✅ | ✅ pd.read_sql() |
프로덕션 코드, 복잡한 쿼리 |
| sqlite3 | Python 내장 | ❌ | ✅ | 로컬 테스트, 간단한 작업 |
| psycopg2 | PostgreSQL 전용 | ❌ | ⚠️ 수동 처리 | PostgreSQL 직접 제어 |
| PyMySQL | MySQL 전용 | ❌ | ⚠️ 수동 처리 | MySQL 직접 제어 |
SQLAlchemy는 데이터베이스 종류에 관계없이 동일한 코드로 작업할 수 있어 실무에서 가장 널리 사용됩니다.
SQLAlchemy 설치 및 데이터베이스 연결
설치
pip install sqlalchemy pandas pymysql psycopg2-binary
엔진 생성
SQLAlchemy의 핵심은 Engine 객체입니다. 데이터베이스 종류별 연결 문자열은 다음과 같습니다.
from sqlalchemy import create_engine
# SQLite (로컬 파일)
engine = create_engine("sqlite:///my_database.db")
# MySQL
engine = create_engine("mysql+pymysql://user:password@host:3306/dbname")
# PostgreSQL
engine = create_engine("postgresql+psycopg2://user:password@host:5432/dbname")
연결 문자열의 구조는 dialect+driver://username:password@host:port/database 형태입니다.
연결 테스트
with engine.connect() as conn:
result = conn.execute(text("SELECT 1"))
print(result.fetchone()) # (1,)
print("연결 성공!")
데이터 추출: SQL → Pandas DataFrame
실전에서 가장 많이 쓰는 패턴은 SQL 쿼리 결과를 바로 DataFrame으로 받는 것입니다.
기본 조회
import pandas as pd
from sqlalchemy import text
query = text("""
SELECT
order_id,
customer_name,
product,
quantity,
price,
order_date
FROM orders
WHERE order_date >= :start_date
ORDER BY order_date DESC
""")
df = pd.read_sql(query, engine, params={"start_date": "2025-01-01"})
print(df.head())
pd.read_sql()에 SQLAlchemy Engine을 전달하면 커넥션 관리를 자동으로 처리합니다. 파라미터 바인딩(:start_date)을 사용하면 SQL 인젝션도 방지할 수 있습니다.
대용량 데이터 처리: chunksize 활용
수백만 건의 데이터를 한 번에 메모리에 올리면 위험합니다. chunksize 옵션으로 나눠서 처리하세요.
chunks = pd.read_sql(
text("SELECT * FROM large_table"),
engine,
chunksize=10000 # 1만 건씩 나눠서 읽기
)
result_list = []
for chunk in chunks:
# 각 청크별 집계
agg = chunk.groupby("category")["amount"].sum()
result_list.append(agg)
final = pd.concat(result_list).groupby(level=0).sum()
Pandas로 데이터 분석하기
DataFrame으로 변환한 뒤에는 Pandas의 강력한 분석 기능을 활용합니다.
실전 분석 예시: 월별 매출 집계
# 날짜 컬럼 변환
df["order_date"] = pd.to_datetime(df["order_date"])
df["month"] = df["order_date"].dt.to_period("M")
# 월별 매출 집계
monthly_sales = (
df.assign(total=df["quantity"] * df["price"])
.groupby("month")["total"]
.agg(["sum", "mean", "count"])
.rename(columns={"sum": "총매출", "mean": "평균주문액", "count": "주문건수"})
)
print(monthly_sales)
분석 결과를 다시 DB에 저장
분석 결과를 데이터베이스 테이블로 저장하는 것도 간단합니다.
monthly_sales.to_sql(
"monthly_sales_report",
engine,
if_exists="replace", # 기존 테이블 대체
index=True
)
if_exists 옵션 |
동작 |
|---|---|
"fail" |
테이블 존재 시 에러 발생 (기본값) |
"replace" |
기존 테이블 삭제 후 재생성 |
"append" |
기존 테이블에 데이터 추가 |
자동화 리포트 만들기
매일 혹은 매주 반복되는 리포트를 자동화하면 업무 시간을 크게 절약할 수 있습니다.
리포트 자동화 스크립트
import pandas as pd
from sqlalchemy import create_engine, text
from datetime import datetime, timedelta
def generate_weekly_report(engine):
"""주간 매출 리포트 생성"""
last_week = datetime.now() - timedelta(days=7)
query = text("""
SELECT
category,
COUNT(*) as order_count,
SUM(quantity * price) as revenue
FROM orders
WHERE order_date >= :since
GROUP BY category
ORDER BY revenue DESC
""")
df = pd.read_sql(query, engine, params={"since": last_week})
# 비율 컬럼 추가
df["비율(%)"] = (df["revenue"] / df["revenue"].sum() * 100).round(1)
# HTML 리포트 생성
html = f"""
<h2>주간 매출 리포트 ({last_week.strftime('%Y-%m-%d')} ~ 현재)</h2>
<p>총 매출: <strong>{df['revenue'].sum():,.0f}원</strong></p>
{df.to_html(index=False)}
"""
# 파일 저장
filename = f"report_{datetime.now().strftime('%Y%m%d')}.html"
with open(filename, "w", encoding="utf-8") as f:
f.write(html)
return filename
# 실행
engine = create_engine("mysql+pymysql://user:pass@host/db")
report_file = generate_weekly_report(engine)
print(f"리포트 생성 완료: {report_file}")
이메일 발송까지 연결
import smtplib
from email.mime.text import MIMEText
def send_report_email(html_content, recipients):
msg = MIMEText(html_content, "html")
msg["Subject"] = f"주간 매출 리포트 - {datetime.now().strftime('%Y-%m-%d')}"
msg["From"] = "report@company.com"
msg["To"] = ", ".join(recipients)
with smtplib.SMTP("smtp.company.com", 587) as server:
server.starttls()
server.login("report@company.com", "password")
server.send_message(msg)
이 스크립트를 cron이나 스케줄러에 등록하면 완전 자동화가 완성됩니다.
# 매주 월요일 오전 9시에 리포트 생성
0 9 * * 1 /usr/bin/python3 /home/user/weekly_report.py
실전 팁 정리
| 상황 | 권장 방법 |
|---|---|
| 단순 SELECT 조회 | pd.read_sql() + text() |
| 대용량 데이터 | chunksize 옵션 활용 |
| 반복 쿼리 | 파라미터 바인딩 (:param) |
| 분석 결과 저장 | df.to_sql() |
| 보안 | 절대 f-string으로 쿼리 조합 금지 |
| 커넥션 관리 | with engine.connect() 컨텍스트 매니저 사용 |
마무리
SQL과 Python 연동의 핵심을 정리하면 다음과 같습니다.
- SQLAlchemy Engine으로 데이터베이스에 연결하면 DB 종류에 관계없이 동일한 코드를 사용할 수 있다
pd.read_sql()로 쿼리 결과를 바로 DataFrame으로 변환하여 분석 워크플로우를 간소화한다- 파라미터 바인딩을 반드시 사용하여 SQL 인젝션을 방지한다
- 대용량 데이터는 chunksize로 나눠 처리하여 메모리 문제를 예방한다
- 분석 결과를
to_sql()로 DB에 저장하고, HTML 리포트 + 이메일 발송까지 자동화하면 반복 업무를 크게 줄일 수 있다
이 워크플로우를 익히면 “데이터 추출 → 분석 → 리포트”라는 데이터 분석의 전체 사이클을 Python 하나로 완성할 수 있습니다.
Did you find this helpful?
☕ Buy me a coffee
Leave a Reply