# Python Data Processing

> Use when working with Python data processing — ETL, data pipelines, type-safe data handling

- Skill: `liaosw97/python-data-processing` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add liaosw97/python-data-processing`
- Raw SKILL.md: https://api.skillmd.com/api/skills/liaosw97/python-data-processing/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Data & Analytics
- Author: liaosw97 (https://skillmd.com/u/liaosw97)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/liaosw97/python-data-processing

---


## 核心原则
- 数据安全第一：始终确保数据完整性和安全性
- 性能优先：使用向量化操作和高效的数据结构
- 可读性至上：编写清晰、可维护的数据处理代码
- 类型安全：使用类型注解提高代码可靠性

## 技术栈
- **语言**：Python 3.10+
- **数据处理**：Pandas, Polars, NumPy
- **数据存储**：PyArrow, Parquet, CSV
- **大数据处理**：Dask, PySpark
- **数据验证**：Pydantic, Pandera
- **并行计算**：concurrent.futures, multiprocessing

## 最佳实践
### 1. 文件操作

```python
# 使用 with 语句安全处理文件
with open('data.csv', 'r', encoding='utf-8') as file:
    data = file.read()

# 使用 pathlib 处理路径
from pathlib import Path

data_dir = Path('data/')
csv_file = data_dir / 'dataset.csv'

if csv_file.exists():
    with csv_file.open('r', encoding='utf-8') as f:
        data = f.read()
```

### 2. 代码风格

```python
# 使用类型注解
from typing import Any
import pandas as pd

def process_data(
    df: pd.DataFrame,
    columns: list[str],
    fill_value: Any = None
) -> pd.DataFrame:
    """处理数据框中的缺失值"""
    return df[columns].fillna(fill_value)

# 使用 PyArrow 进行高效数据读取
import pyarrow.parquet as pq

table = pq.read_table('data.parquet')
df = table.to_pandas()

# 使用 Polars 进行高性能数据处理
import polars as pl

df = pl.read_csv('data.csv')
result = (
    df
    .filter(pl.col('status') == 'active')
    .group_by('category')
    .agg(pl.col('value').mean())
)
```

### 3. 内存管理

```python
# 分块读取大文件
chunk_size = 100000
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
    process_chunk(chunk)

# 使用适当的数据类型减少内存占用
df = df.astype({
    'id': 'int32',
    'category': 'category',
    'flag': 'bool'
})
```

### 4. 数据验证

```python
import pandera as pa

schema = pa.DataFrameSchema({
    'id': pa.Column(int, checks=pa.Check.ge(0)),
    'name': pa.Column(str, nullable=False),
    'value': pa.Column(float, checks=pa.Check.lt(100)),
})

validated_df = schema.validate(df)
```

### 5. 性能优化

```python
# 向量化操作
df['total'] = df['quantity'] * df['price']

# 使用 apply 时优化
df['processed'] = df['text'].str.lower().str.strip()

# 并行处理
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(process_item, items))
```

### 6. 数据管道

```python
from functools import partial

def load_data(path: Path) -> pd.DataFrame:
    return pd.read_csv(path)

def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    return df.dropna().drop_duplicates()

def transform_data(df: pd.DataFrame, column: str) -> pd.DataFrame:
    df[column] = df[column].astype('category')
    return df

# 管道式处理
pipeline = [
    load_data,
    clean_data,
    partial(transform_data, column='category')
]

result = df
for step in pipeline:
    result = step(result)
```

## 关键约定
1. **命名约定**
   - DataFrame 变量使用 `_df` 后缀
   - Series 变量使用 `_series` 后缀
   - 路径变量使用 `Path` 类型

2. **数据处理约定**
   - 始终处理缺失值和异常值
   - 保留原始数据，创建处理后的副本
   - 记录数据转换步骤

3. **文件格式约定**
   - 优先使用 Parquet 格式存储大数据
   - CSV 文件使用 UTF-8 编码
   - 大文件使用压缩（gzip, snappy）

4. **项目结构**

```
data-project/
├── data/
│   ├── raw/
│   ├── processed/
│   └── output/
├── src/
│   ├── __init__.py
│   ├── extract.py
│   ├── transform.py
│   └── load.py
├── notebooks/
├── tests/
│   └── test_pipeline.py
├── pyproject.toml
└── README.md
```

## 测试
- 使用 pytest 进行单元测试
- 测试数据处理函数的边界情况
- 使用 fixtures 准备测试数据
- 测试数据验证逻辑

## 文档
- 为数据处理函数添加 docstrings
- 记录数据模式和字段含义
- 维护数据字典