pandas 라이브러리는 Python의 주요 데이터 분석 도구입니다. pandas와 mssql-python 드라이버를 결합하면 다음과 같은 기능이 있습니다:
- SQL 쿼리 결과를 DataFrames에 직접 불러오세요.
- DataFrame을 Microsoft SQL에 효율적으로 다시 쓰세요.
- ETL 작업을 수행하세요.
- 데이터 파이프라인을 만드세요.
이 글의 예시들은 Production.Product의 테이블과 다른 테이블들을 조회합니다. 데이터를 작성하는 예시는 샘플 데이터를 수정하지 않기 위해 임시 테이블을 사용합니다.
분석 예시(Sales.SalesOrderHeader, Sales.SalesOrderDetail, Production.ProductSubcategory)에서 언급된 다른 표들은 AdventureWorks의 일부입니다. 이러한 패턴을 적용할 때는 자체 표로 대체하세요.
데이터를 데이터프레임에 읽기
mssql-python 드라이버는 행을 Python 객체로 반환하며, cursor.description에서 열 이름을 읽고 fetchall()에서 행 값을 읽어 pandas DataFrame으로 변환합니다. 이 섹션의 헬퍼는 변환 과정을 재사용 가능한 패턴으로 감싸는 역할을 합니다.
DataFrame에 대한 기본 쿼리
이 함수는 매개변수화된 쿼리 를 실행하고 전체 결과 집합으로부터 데이터프레임을 만듭니다. 메모리에 무리 없이 들어가는 결과 집합에 효과적입니다.
import pandas as pd
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
def query_to_dataframe(cursor, query: str, params: dict = None) -> pd.DataFrame:
"""Execute query and return results as DataFrame."""
cursor.execute(query, params or {})
# cursor.description is a list of tuples, one per column.
# Each tuple's first element is the column name.
columns = [col[0] for col in cursor.description]
# Fetch all rows
rows = cursor.fetchall()
# Convert to DataFrame
data = [tuple(row) for row in rows]
return pd.DataFrame(data, columns=columns)
# Usage: %(cat)s is a parameterized placeholder. The driver safely substitutes
# the value from the dict, which prevents SQL injection.
df = query_to_dataframe(cursor, "SELECT * FROM Production.Product WHERE ProductSubcategoryID = %(cat)s", {"cat": 5})
print(df.head())
비고
연결 문자열에서 Authentication=ActiveDirectoryDefault을(를) 사용하는 경우 드라이버는 DefaultAzureCredential을(를) 사용하며, 이는 여러 자격 증명 공급자를 순차적으로 시도합니다. 첫 번째 연결은 SDK가 작동하는 공급자를 찾을 때까지 체인을 따라 걸어 다니기 때문에 느릴 수 있습니다. 운영 환경에서는 어떤 자격 증명 유형을 사용하는지 알면, 체인 워크를 피하기 위해 직접 지정하세요(예: ActiveDirectoryMSI 관리 신원). 자세한 내용은 Microsoft Entra 인증을 참조하세요.
대규모 데이터셋 스트리밍
수백만 행이 있는 테이블에서는 모든 것을 한꺼번에 로드하면 메모리가 소진될 수 있습니다. 청크 방식은 fetchmany()를 사용해 행을 배치 단위로 가져오고 그 결과를 연결하여, 최대 메모리 사용량이 전체 결과 집합이 아니라 chunksize에 비례하도록 유지합니다.
def query_to_dataframe_chunked(cursor, query: str, params: dict = None,
chunksize: int = 10000) -> pd.DataFrame:
"""Load large query results in chunks for memory efficiency."""
cursor.execute(query, params or {})
columns = [col[0] for col in cursor.description]
chunks = []
while True:
rows = cursor.fetchmany(chunksize)
if not rows:
break
data = [tuple(row) for row in rows]
chunks.append(pd.DataFrame(data, columns=columns))
return pd.concat(chunks, ignore_index=True) if chunks else pd.DataFrame(columns=columns)
# Usage for large tables
df = query_to_dataframe_chunked(cursor, "SELECT * FROM Production.TransactionHistory", chunksize=50000)
대규모 데이터셋용 생성기
전체 결과를 메모리에 저장하지 않고 점진적으로 데이터를 처리해야 할 때는 생성기를 사용하세요. 각 yield는 하나의 DataFrame 청크를 생성하며, 다음 청크를 가져오기 전에 이를 처리하고 폐기할 수 있습니다.
def query_to_dataframe_generator(cursor, query: str, params: dict = None,
chunksize: int = 10000):
"""Yield DataFrame chunks for processing without loading all data."""
cursor.execute(query, params or {})
columns = [col[0] for col in cursor.description]
while True:
rows = cursor.fetchmany(chunksize)
if not rows:
break
data = [tuple(row) for row in rows]
yield pd.DataFrame(data, columns=columns)
# Process chunks without loading entire dataset
huge_query = """
SELECT * FROM Production.TransactionHistory
UNION ALL SELECT * FROM Production.TransactionHistory
UNION ALL SELECT * FROM Production.TransactionHistory
"""
for chunk_df in query_to_dataframe_generator(cursor, huge_query):
# Process each chunk, then discard it before the next fetch
print(f"Processing chunk of {len(chunk_df)} rows")
total_cost = chunk_df["ActualCost"].sum()
print(f"Chunk total cost: {total_cost}")
Microsoft SQL에 DataFrames 작성하기
SQL 주입을 방지하기 위한 인용 식별자
테이블 이름과 열명은 SQL에서 쿼리 매개변수로 전달할 수 없습니다. 동적 식별자를 가진 SQL 문장을 만들 때는 각 이름을 대괄호로 감싸고 임베디드 ] 문자는 제거하여 SQL 인젝션을 방지하세요.
def quote_id(identifier: str) -> str:
"""Quote a Microsoft SQL identifier to prevent SQL injection.
Wraps the name in square brackets and escapes any embedded ] characters.
Raises ValueError if the identifier is empty or contains null bytes.
"""
if not identifier or "\x00" in identifier:
raise ValueError(f"Invalid identifier: {identifier!r}")
escaped = identifier.replace("]", "]]")
return f"[{escaped}]"
이 섹션의 헬퍼 함수는 생성된 SQL의 모든 테이블 및 열 이름에 quote_id()를 사용합니다.
데이터프레임 행 삽입
가장 간단한 방법은 DataFrame 행을 반복하여 각 행마다 하나씩 INSERT 발행하는 방식입니다. 간단한 방법은 작은 데이터프레임에는 효과적이지만, 각 행이 서버로 별도의 왕복 경로를 필요로 하기 때문에 큰 볼륨에서는 느립니다.
def dataframe_to_sql(cursor, conn, df: pd.DataFrame, table: str,
if_exists: str = "append") -> int:
"""Write DataFrame to Microsoft SQL table."""
if if_exists == "replace":
cursor.execute(f"TRUNCATE TABLE {quote_id(table)}")
columns = df.columns.tolist()
placeholders = ", ".join([f"%({col})s" for col in columns])
col_list = ", ".join([quote_id(col) for col in columns])
query = f"INSERT INTO {quote_id(table)} ({col_list}) VALUES ({placeholders})"
rows_inserted = 0
for _, row in df.iterrows():
params = {col: (None if pd.isna(val) else val) for col, val in row.items()}
cursor.execute(query, params)
rows_inserted += 1
conn.commit()
return rows_inserted
# Usage
cursor.execute("""
CREATE TABLE #Products (
Name NVARCHAR(100),
ListPrice DECIMAL(10,2),
ProductSubcategoryID INT
)
""")
df = pd.DataFrame({
"Name": ["Product A", "Product B"],
"ListPrice": [29.99, 49.99],
"ProductSubcategoryID": [1, 2]
})
rows = dataframe_to_sql(cursor, conn, df, "#Products")
print(f"Inserted {rows} rows")
BCP가 포함된 대량 인서트 (대용량 데이터프레임에 권장)
대용량 데이터프레임의 경우, 드라이버 bulkcopy() 방식을 사용하는데, 이는 Microsoft SQL이 사용하는 네이티브 와이어 프로토콜인 TDS(Tabular Data Stream) 프로토콜을 통해 행을 대량으로 전송합니다. 이 방식은 왕복 통신을 최소화하므로 한 행씩 삽입하는 것보다 빠릅니다.
def dataframe_to_sql_bulk(conn, df: pd.DataFrame, table: str) -> int:
"""Bulk insert DataFrame using BCP for better performance."""
# Convert DataFrame to list of tuples, handling NaN
rows = []
for _, row in df.iterrows():
row_data = tuple(None if pd.isna(v) else v for v in row)
rows.append(row_data)
cursor = conn.cursor()
result = cursor.bulkcopy(table, rows)
conn.commit()
return result["rows_copied"]
# Usage
cursor.execute("CREATE TABLE ##PandasProducts (Name NVARCHAR(50), ListPrice DECIMAL(10,2), ProductSubcategoryID INT)")
conn.commit()
df = pd.DataFrame({
"Name": ["Product A", "Product B", "Product C"],
"ListPrice": [29.99, 49.99, 19.99],
"ProductSubcategoryID": [1, 2, 1]
})
rows = dataframe_to_sql_bulk(conn, df, "##PandasProducts")
DataFrame에서 기존 행을 업데이트하기
테이블에 이미 존재하는 행을 업데이트하려면 DataFrame을 반복하여 매개변수화된 UPDATE 문장을 발행합니다.
key_column은(는) 업데이트할 행을 결정합니다.
def update_from_dataframe(cursor, conn, df: pd.DataFrame, table: str,
key_column: str) -> int:
"""Update existing rows based on key column."""
columns = [col for col in df.columns if col != key_column]
set_clause = ", ".join([f"{quote_id(col)} = %({col})s" for col in columns])
query = f"UPDATE {quote_id(table)} SET {set_clause} WHERE {quote_id(key_column)} = %({key_column})s"
rows_updated = 0
for _, row in df.iterrows():
params = {col: (None if pd.isna(val) else val) for col, val in row.items()}
cursor.execute(query, params)
rows_updated += cursor.rowcount
conn.commit()
return rows_updated
# Usage
cursor.execute("""
CREATE TABLE #ProductPrices (
ProductID INT PRIMARY KEY,
ListPrice DECIMAL(10,2)
);
INSERT INTO #ProductPrices VALUES (1, 29.99), (2, 49.99), (3, 19.99);
""")
conn.commit()
df_updates = pd.DataFrame({
"ProductID": [1, 2, 3],
"ListPrice": [31.99, 52.99, 21.99]
})
updated = update_from_dataframe(cursor, conn, df_updates, "#ProductPrices", "ProductID")
업서트(병합) 패턴
어떤 행은 새로워 있고 다른 행은 이미 존재할 수 있다면, SQL MERGE 문장을 사용해 한 번의 연산으로 삽입하거나 업데이트하세요.
MERGE 키 열을 사용하여 각 들어오는 행을 목표 테이블과 비교합니다. 일치하는 것이 발견되면 업데이트됩니다; 그렇지 않으면 삽입됩니다.
MERGE 별도로 존재 여부를 확인하는 것을 피합니다.
def upsert_from_dataframe(cursor, conn, df: pd.DataFrame, table: str,
key_columns: list[str]) -> int:
"""Insert or update rows based on key columns. Returns total rows affected."""
all_columns = df.columns.tolist()
value_columns = [c for c in all_columns if c not in key_columns]
total_affected = 0
for _, row in df.iterrows():
params = {col: (None if pd.isna(val) else val) for col, val in row.items()}
# Build MERGE statement with quoted identifiers
key_match = " AND ".join([f"t.{quote_id(k)} = s.{quote_id(k)}" for k in key_columns])
update_set = ", ".join([f"{quote_id(c)} = s.{quote_id(c)}" for c in value_columns])
all_cols = ", ".join([quote_id(c) for c in all_columns])
all_vals = ", ".join([f"%({c})s" for c in all_columns])
cursor.execute(f"""
MERGE {quote_id(table)} AS t
USING (SELECT {', '.join([f'%({c})s AS {quote_id(c)}' for c in all_columns])}) AS s
ON {key_match}
WHEN MATCHED THEN UPDATE SET {update_set}
WHEN NOT MATCHED THEN INSERT ({all_cols}) VALUES ({all_vals});
""", params)
total_affected += cursor.rowcount
conn.commit()
return total_affected
데이터 분석 패턴
다음 예시들은 Microsoft SQL 쿼리와 판다스 변환을 결합한 일반적인 분석 작업을 보여줍니다.
DataFrame에 대한 집계 쿼리
def get_sales_summary(cursor) -> pd.DataFrame:
"""Get sales summary by category."""
return query_to_dataframe(cursor, """
SELECT
pc.Name AS CategoryName,
COUNT(*) AS ProductCount,
AVG(p.ListPrice) AS AvgPrice,
MIN(p.ListPrice) AS MinPrice,
MAX(p.ListPrice) AS MaxPrice
FROM Production.Product p
JOIN Production.ProductSubcategory pc ON p.ProductSubcategoryID = pc.ProductSubcategoryID
GROUP BY pc.Name
ORDER BY ProductCount DESC
""")
df = get_sales_summary(cursor)
print(df.to_string())
시계열 데이터
pandas 날짜 인덱싱과 리샘플링을 사용해 Microsoft SQL의 시계열 데이터를 활용하세요. 롤링 평균이나 재샘플링 같은 연산을 가능하게 하려면 날짜 열을 DataFrame 인덱스로 설정하세요.
def get_daily_sales(cursor, start_date: str, end_date: str) -> pd.DataFrame:
"""Get daily sales time series."""
df = query_to_dataframe(cursor, """
SELECT
CAST(OrderDate AS DATE) AS Date,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS Revenue
FROM Sales.SalesOrderHeader
WHERE OrderDate BETWEEN %(start)s AND %(end)s
GROUP BY CAST(OrderDate AS DATE)
ORDER BY Date
""", {"start": start_date, "end": end_date})
# Set date as index for time series operations
df["Date"] = pd.to_datetime(df["Date"])
df.set_index("Date", inplace=True)
return df
# Usage
sales_df = get_daily_sales(cursor, "2024-01-01", "2024-12-31")
# Resample to weekly
weekly = sales_df.resample("W").sum()
# Calculate rolling average
sales_df["RollingAvg"] = sales_df["Revenue"].rolling(window=7).mean()
SQL 데이터에서 얻은 피벗 테이블
피벗 테이블은 데이터를 행에서 행렬 형식으로 재구성합니다. 연도, 월, 카테고리 같은 차원별로 재구성하려면 Microsoft SQL에서 원시 데이터를 가져와서 .을 사용하세요pivot_table().
def get_sales_pivot(cursor) -> pd.DataFrame:
"""Get sales data and create pivot table."""
df = query_to_dataframe(cursor, """
SELECT
YEAR(soh.OrderDate) AS Year,
MONTH(soh.OrderDate) AS Month,
pc.Name AS CategoryName,
SUM(sod.OrderQty * sod.UnitPrice) AS Revenue
FROM Sales.SalesOrderHeader soh
JOIN Sales.SalesOrderDetail sod ON soh.SalesOrderID = sod.SalesOrderID
JOIN Production.Product p ON sod.ProductID = p.ProductID
JOIN Production.ProductSubcategory pc ON p.ProductSubcategoryID = pc.ProductSubcategoryID
GROUP BY YEAR(soh.OrderDate), MONTH(soh.OrderDate), pc.Name
""")
# Create pivot table
pivot = df.pivot_table(
values="Revenue",
index=["Year", "Month"],
columns="CategoryName",
aggfunc="sum",
fill_value=0
)
return pivot
pivot_df = get_sales_pivot(cursor)
print(pivot_df)
ETL 패턴
추출, 변환, 로드 파이프라인을 구축하려면 Microsoft SQL 쿼리와 pandas 변환을 결합하여 추출, 변환, 로드 파이프라인을 구축할 수 있습니다. 드라이버는 추출과 적재를 담당하고, 판다스는 변환 단계를 담당합니다.
추출, 변환, 로드
이 예시는 활성 고객 데이터를 추출하고, 비즈니스 규칙을 세분화된 고객에게 적용하며, 결과를 목적지 테이블에 로드합니다.
def etl_pipeline(source_cursor, dest_cursor, dest_conn):
"""Simple ETL pipeline with pandas."""
# Extract
df = query_to_dataframe(source_cursor, """
SELECT
c.CustomerID,
COUNT(soh.SalesOrderID) AS OrderCount,
SUM(soh.TotalDue) AS TotalSpent
FROM Sales.Customer c
JOIN Sales.SalesOrderHeader soh ON c.CustomerID = soh.CustomerID
WHERE soh.OrderDate > DATEADD(YEAR, -1, GETDATE())
GROUP BY c.CustomerID
""")
# Transform
df["CustomerSegment"] = pd.cut(
df["TotalSpent"],
bins=[0, 100, 500, 1000, float("inf")],
labels=["Bronze", "Silver", "Gold", "Platinum"]
)
df["AvgOrderValue"] = df["TotalSpent"] / df["OrderCount"].replace(0, 1)
df["IsHighValue"] = df["TotalSpent"] > 500
# Load
dataframe_to_sql_bulk(dest_conn, df[["CustomerID", "CustomerSegment", "AvgOrderValue", "IsHighValue"]],
"#CustomerAnalytics")
return len(df)
점진적 부하 패턴
진행 중인 데이터 파이프라인의 경우, 마지막 실행 이후 변경된 레코드만 불러오세요. 이 방법은 목적지 테이블에서 최대 타임스탬프를 쿼리한 후, 소스에서 새 레코드만 가져옵니다.
def incremental_load(cursor, conn, source_table: str, dest_table: str,
timestamp_col: str) -> int:
"""Load only new/changed records based on timestamp."""
# Get last loaded timestamp
cursor.execute(f"SELECT MAX({quote_id(timestamp_col)}) FROM {quote_id(dest_table)}")
last_loaded = cursor.fetchval()
# Build query for new records
if last_loaded:
df = query_to_dataframe(cursor, f"""
SELECT * FROM {quote_id(source_table)}
WHERE {quote_id(timestamp_col)} > %(last)s
""", {"last": last_loaded})
else:
df = query_to_dataframe(cursor, f"SELECT * FROM {quote_id(source_table)}")
if df.empty:
return 0
# Load new records
return dataframe_to_sql_bulk(conn, df, dest_table)
성능 팁
적절한 데이터 형식 사용
Pandas는 기본적으로 숫자에 대해 64비트 타입을 사용하는데, 작은 타입으로도 충분하지만 메모리를 낭비합니다. 정수와 플로트 수를 다운캐스트하고, 저카디널리티 문자열 열을 범주형으로 변환하면 메모리 사용량을 크게 줄일 수 있습니다.
def optimize_dataframe_types(df: pd.DataFrame) -> pd.DataFrame:
"""Optimize DataFrame memory usage."""
for col in df.columns:
col_type = df[col].dtype
if col_type == "int64":
# Downcast integers
df[col] = pd.to_numeric(df[col], downcast="integer")
elif col_type == "float64":
# Downcast floats
df[col] = pd.to_numeric(df[col], downcast="float")
elif col_type == "object":
# Convert to category if low cardinality
num_unique = df[col].nunique()
if num_unique / len(df) < 0.5:
df[col] = df[col].astype("category")
return df
무거운 작업에는 SQL 사용
Microsoft SQL은 모든 원시 데이터를 직접 끌어와서 Python으로 로컬로 처리하는 것보다 집계, 필터링, 조인에 더 빠릅니다. 가능한 한 Microsoft SQL이 큰 역할을 맡기고, 필요한 데이터만 네트워크를 통해 이동시키며, 분석과 변환은 Python에서 더 편리한 pandas를 사용하세요.
# Avoid: pulling all rows over the wire to aggregate locally in pandas
df_all = query_to_dataframe(cursor, "SELECT * FROM Production.Product") # transfers entire table
summary = df_all.groupby("Color").agg({"ListPrice": "sum"}) # aggregation that SQL can do faster
# Better: push the aggregation into SQL and transfer only the summary
df = query_to_dataframe(cursor, """
SELECT Color, SUM(ListPrice) AS TotalPrice
FROM Production.Product
WHERE Color IS NOT NULL
GROUP BY Color
""")
일괄 쓰기
단일 대용량 인서트가 너무 큰 데이터프레임의 경우, 작업을 배치로 나누어 진행 상황을 추적하세요.
def batch_insert(cursor, conn, df: pd.DataFrame, table: str, batch_size: int = 1000):
"""Insert in batches with progress tracking."""
total = len(df)
for i in range(0, total, batch_size):
batch = df.iloc[i:i + batch_size]
dataframe_to_sql(cursor, conn, batch, table)
print(f"Inserted {min(i + batch_size, total)}/{total}")