다중 파일 데이터셋 다루기
Polars로 데이터 파이프라인 확장 및 최적화하기
Liam Brannigan
Data Scientist & Polars Contributor
일별 311 파일
311_requests_daily_csv/
311_requests_2025-12-31.csv
311_requests_2026-01-01.csv
311_requests_2026-01-02.csv
311_requests_2026-01-03.csv
311_requests_2026-01-04.csv
다수의 CSV 파일 스캔
pl.scan_csv(
)
다수의 CSV 파일 스캔
pl.scan_csv(
"311_requests_daily_csv/311_requests_2026*.csv",
try_parse_dates=True,
)
다수의 CSV 파일 스캔
pl.scan_csv(
"311_requests_daily_csv/311_requests_2026*.csv",
try_parse_dates=True,
).select("REPORTED_AT", "STATUS", "WARD").collect()
shape: (4, 3)
| REPORTED_AT | STATUS | WARD |
| --- | --- | --- |
| datetime[us] | str | i64 |
|---------------------|-------------|------|
| 2026-01-01 07:10:00 | In Progress | 46 |
| 2026-01-01 08:25:00 | Open | 46 |
| 2026-01-04 07:45:00 | Open | 12 |
하이브 파티션 쓰기
pl.read_csv(
"311_Service_Requests.csv",
try_parse_dates=True,
).write_parquet(
)
하이브 파티션 쓰기
pl.read_csv(
"311_Service_Requests.csv",
try_parse_dates=True,
).write_parquet(
"311_requests_parquet/",
)
하이브 파티션 쓰기
pl.read_csv(
"311_Service_Requests.csv",
try_parse_dates=True,
).write_parquet(
"311_requests_parquet/",
partition_by=["CREATED_DATE"],
)
파티션된 데이터셋
311_requests_parquet/
CREATED_DATE=2025-12-31/
data.parquet
파티션된 데이터셋
311_requests_parquet/
CREATED_DATE=2025-12-31/
data.parquet
CREATED_DATE=2026-01-01/
data.parquet
하이브 파티션 데이터셋 스캔
requests = pl.scan_parquet(
"311_requests_parquet/",
hive_partitioning=True,
)
$$
- 파티션 값 → 일반 열
파티션 인식 쿼리
requests.filter(
pl.col("CREATED_DATE") >= date(2026, 1, 1)
)
파티션 인식 쿼리
requests.filter(
pl.col("CREATED_DATE") >= date(2026, 1, 1)
).group_by("DEPARTMENT").len().collect()
shape: (2, 2)
| DEPARTMENT | len |
| --- | --- |
| str | u32 |
|----------------|-----|
| Transportation | 208 |
| Sanitation | 175 |
파티션 미인식 쿼리!
requests.filter(
pl.col("YEAR") >= 2026
)
$$
- Polars는 파티션 열을 직접 필터링합니다 💡
- 파생 열에 필터링하면 최적화가 작동하지 않습니다 💡
스키마 드리프트 문제
# 2026-01-01 daily file
REPORTED_AT,STATUS,WARD,BLOCK_CODE,REPORT_TEXT
2026-01-01 07:10:00,In Progress,46,0021,...
# 2026-01-02 daily file
REPORTED_AT,STATUS,WARD,REPORT_TEXT
2026-01-02 09:05:00,Open,29,...
누락 열 삽입
requests = pl.scan_csv(
"311_requests_daily_csv/311_requests_2026*.csv",
try_parse_dates=True,
missing_columns="insert",
)
통합 스키마 확인
requests.select(
"REPORTED_AT", "STATUS", "WARD", "BLOCK_CODE"
).collect()
shape: (8, 4)
| REPORTED_AT | STATUS | WARD | BLOCK_CODE |
| --- | --- | --- | --- |
| datetime[us] | str | i64 | i64 |
|---------------------|-------------|------|------------|
| 2026-01-01 07:10:00 | In Progress | 46 | 21 |
| 2026-01-01 08:25:00 | Open | 46 | 21 |
| 2026-01-02 09:05:00 | Open | 29 | null |
| 2026-01-02 10:42:00 | Completed | 29 | null |
데이터 타입 불일치 문제
# 2025-12-31 daily file
WARD
35
# 2026-01-03 daily file
WARD
"44"
서로 다른 데이터 타입으로 두 번 스캔
ward_int = pl.scan_csv(
"311_requests_daily_csv/311_requests_2025-12-31.csv",
try_parse_dates=True,
)
ward_str = pl.scan_csv(
"311_requests_daily_csv/311_requests_2026-01-03.csv",
try_parse_dates=True,
schema_overrides={"WARD": pl.String},
)
vertical_relaxed로 결합하기
combined = pl.concat(
[ward_int, ward_str],
how="vertical_relaxed",
)
$$
- 수퍼타입: 두 입력의 값을 모두 담을 수 있는 데이터 타입
- 여기서 String은 Int64와 String 모두 포함 가능
relaxed concat 결과 확인
combined.collect()
shape: (4, 5)
| REPORTED_AT | STATUS | WARD | BLOCK_CODE | REPORT_TEXT |
| --- | --- | --- | --- | --- |
| datetime[us] | str | str | i64 | str |
|---------------------|-----------|------|------------|----------------------------------|
| 2025-12-31 16:20:00 | Open | 35 | 17 | Massive pothole getting worse |
| 2025-12-31 18:05:00 | Open | 35 | 17 | Cyclists forced into traffic |
| 2026-01-03 07:35:00 | Open | 44 | 12 | How is this still not repaired |
| 2026-01-03 08:15:00 | Completed | 44 | 12 | Every bus hit shakes the vehicle |
연습해 봅시다!
Polars로 데이터 파이프라인 확장 및 최적화하기
