Python 数据处理实战：Pandas 核心 API 速查 + 大文件分块读取 + 内存优化技巧

1
import pandas as pd
2
import numpy as np
3

4
# 从字典创建
5
df = pd.DataFrame({
6
    'name':   ['Alice', 'Bob', 'Charlie', 'Diana'],
7
    'age':    [25, 30, 35, 28],
8
    'city':   ['北京', '上海', '北京', '广州'],
9
    'salary': [8000, 12000, 15000, 9500],
10
    'score':  [88.5, 92.0, 78.3, 95.1],
11
})
12

13
# 读取 CSV
14
df = pd.read_csv('data.csv', encoding='utf-8')
15

16
# 读取 Excel（需要 openpyxl）
17
df = pd.read_excel('data.xlsx', sheet_name='Sheet1')
18

19
# 读取 JSON
20
df = pd.read_json('data.json', orient='records')
21

22
# 读取数据库（需要 sqlalchemy）
23
from sqlalchemy import create_engine
24
engine = create_engine('sqlite:///data.db')
25
df = pd.read_sql('SELECT * FROM users LIMIT 1000', engine)

1
df.shape          # (行数, 列数)
2
df.dtypes         # 各列数据类型
3
df.info()         # 综合信息：类型 + 非空数量 + 内存占用
4
df.describe()     # 数值列统计（均值/标准差/分位数）
5
df.head(5)        # 前 5 行
6
df.tail(3)        # 后 3 行
7
df.sample(10)     # 随机抽 10 行
8

9
df.isnull().sum()           # 各列缺失值数量
10
df.duplicated().sum()       # 重复行数量
11
df['city'].value_counts()   # 某列的频次统计
12
df['salary'].describe()     # 单列统计

1
# ── 按列选取 ──────────────────────────────────────────────────
2
df['name']                  # 单列 → Series
3
df[['name', 'salary']]      # 多列 → DataFrame
4

5
# ── 按行选取 ──────────────────────────────────────────────────
6
df.loc[0]                   # 按索引标签选行
7
df.loc[0:2]                 # 按标签切片（包含末端！）
8
df.iloc[0]                  # 按位置选行
9
df.iloc[0:3]                # 按位置切片（不包含末端）
10
df.iloc[[0, 2, 4]]          # 按位置列表选行
11

12
# ── 行列同时选取 ──────────────────────────────────────────────
13
df.loc[0:2, ['name', 'salary']]        # 按标签：前3行的 name/salary 列
14
df.iloc[0:3, [0, 3]]                   # 按位置：前3行的第0和第3列
15

16
# ── 条件过滤（最常用）──────────────────────────────────────────
17
df[df['age'] > 28]                          # 单条件
18
df[(df['age'] > 25) & (df['city'] == '北京')]  # 多条件（用 & 而非 and）
19
df[df['city'].isin(['北京', '上海'])]         # isin 过滤
20
df[df['name'].str.contains('li', case=False)]  # 字符串包含
21

22
# query 语法（更简洁）
23
df.query('age > 28 and city == "北京"')
24
df.query('salary > @threshold', threshold=10000)  # 引用外部变量

1
# ── 缺失值处理 ────────────────────────────────────────────────
2
df.dropna()                         # 删除含任意缺失值的行
3
df.dropna(subset=['salary'])        # 只删除 salary 为空的行
4
df.dropna(thresh=3)                 # 至少有 3 个非空值才保留
5
df.fillna(0)                        # 用 0 填充所有缺失值
6
df['salary'].fillna(df['salary'].mean())  # 用均值填充
7
df['city'].fillna(method='ffill')   # 前向填充（用上一个非空值）
8

9
# ── 重复值处理 ────────────────────────────────────────────────
10
df.drop_duplicates()                # 删除完全重复的行
11
df.drop_duplicates(subset=['name', 'city'])  # 按指定列判断重复
12

13
# ── 类型转换 ──────────────────────────────────────────────────
14
df['age'] = df['age'].astype(int)
15
df['date'] = pd.to_datetime(df['date'])  # 字符串转日期
16
df['salary'] = pd.to_numeric(df['salary'], errors='coerce')  # 转数值，错误置 NaN
17

18
# ── 字符串操作（str accessor）────────────────────────────────
19
df['name'].str.upper()              # 转大写
20
df['name'].str.strip()              # 去除首尾空格
21
df['city'].str.replace('北京', 'Beijing')  # 替换
22
df['email'].str.split('@').str[0]  # 按分隔符拆分取第一部分
23
df['name'].str.len()               # 字符串长度

1
df.sort_values('salary', ascending=False)                  # 按薪资降序
2
df.sort_values(['city', 'salary'], ascending=[True, False]) # 多列排序
3
df['salary'].rank(method='dense')   # 排名（dense = 无间断）
4
df.nlargest(5, 'salary')            # 薪资前5名
5
df.nsmallest(3, 'score')            # 分数最低3名

1
# ── 基础分组聚合 ──────────────────────────────────────────────
2
df.groupby('city')['salary'].mean()         # 各城市平均薪资
3
df.groupby('city')['salary'].agg(['mean', 'max', 'min', 'count'])
4

5
# ── 多列分组 ──────────────────────────────────────────────────
6
df.groupby(['city', 'department'])['salary'].mean()
7

8
# ── 多个聚合指标 ──────────────────────────────────────────────
9
result = df.groupby('city').agg(
10
    avg_salary   = ('salary', 'mean'),
11
    max_salary   = ('salary', 'max'),
12
    employee_cnt = ('name', 'count'),
13
    avg_score    = ('score', 'mean'),
14
).round(2)
15

16
# ── 自定义聚合函数 ────────────────────────────────────────────
17
# 计算每个城市薪资的"极差"（最大值 - 最小值）
18
df.groupby('city')['salary'].agg(lambda x: x.max() - x.min())
19

20
# ── transform：保持原 DataFrame 行数（不合并）──────────────────
21
# 给每行加一列"所在城市平均薪资"
22
df['city_avg_salary'] = df.groupby('city')['salary'].transform('mean')
23

24
# ── filter：过滤整个分组 ──────────────────────────────────────
25
# 只保留城市平均薪资 > 10000 的行
26
df.groupby('city').filter(lambda x: x['salary'].mean() > 10000)

1
# 假设有两个 DataFrame
2
employees = pd.DataFrame({
3
    'emp_id': [1, 2, 3, 4],
4
    'name':   ['Alice', 'Bob', 'Charlie', 'Diana'],
5
    'dept_id': [10, 20, 10, 30],
6
})
7
departments = pd.DataFrame({
8
    'dept_id':   [10, 20, 40],
9
    'dept_name': ['Engineering', 'Marketing', 'HR'],
10
})
11

12
# ── 内连接（交集，类似 SQL INNER JOIN）──────────────────────
13
pd.merge(employees, departments, on='dept_id', how='inner')
14
# 结果：只保留两个表都有的 dept_id（10, 20）
15

16
# ── 左连接（保留左表全部行）────────────────────────────────────
17
pd.merge(employees, departments, on='dept_id', how='left')
18
# 结果：Diana 的 dept_id=30 在 departments 中不存在，dept_name 为 NaN
19

20
# ── 外连接（并集）────────────────────────────────────────────
21
pd.merge(employees, departments, on='dept_id', how='outer')
22

23
# ── 不同列名的连接 ────────────────────────────────────────────
24
pd.merge(employees, departments,
25
         left_on='dept_id', right_on='dept_id')  # 明确指定
26

27
# ── 多键连接 ──────────────────────────────────────────────────
28
pd.merge(df1, df2, on=['city', 'year'])
29

30
# ── concat：纵向/横向拼接 ─────────────────────────────────────
31
pd.concat([df1, df2], axis=0, ignore_index=True)   # 纵向拼接（叠加行）
32
pd.concat([df1, df2], axis=1)                       # 横向拼接（叠加列）

1
import pandas as pd
2

3
# 示例数据：销售记录
4
sales = pd.DataFrame({
5
    'month':    ['Jan', 'Jan', 'Feb', 'Feb', 'Mar', 'Mar'],
6
    'city':     ['北京', '上海', '北京', '上海', '北京', '上海'],
7
    'product':  ['A', 'B', 'A', 'B', 'A', 'B'],
8
    'revenue':  [1000, 1500, 1200, 1800, 900, 2000],
9
    'quantity': [10, 15, 12, 18, 9, 20],
10
})
11

12
# ── 基础透视：各月各城市的总收入 ─────────────────────────────
13
pd.pivot_table(
14
    sales,
15
    values='revenue',       # 汇总的值
16
    index='month',          # 行（纵轴维度）
17
    columns='city',         # 列（横轴维度）
18
    aggfunc='sum',          # 聚合方式
19
    fill_value=0,           # 空值填充为 0
20
)
21
# 输出：
22
# city   北京    上海
23
# month
24
# Feb    1200   1800
25
# Jan    1000   1500
26
# Mar     900   2000
27

28
# ── 多个聚合指标 ──────────────────────────────────────────────
29
pd.pivot_table(
30
    sales,
31
    values=['revenue', 'quantity'],
32
    index=['month', 'city'],
33
    aggfunc={'revenue': 'sum', 'quantity': 'mean'},
34
    margins=True,           # 添加合计行/列
35
    margins_name='总计',
36
)

1
# 逐行运算（axis=1）
2
df['bonus'] = df.apply(
3
    lambda row: row['salary'] * 0.2 if row['score'] >= 90 else row['salary'] * 0.1,
4
    axis=1
5
)
6

7
# 逐列运算（axis=0，用于统计）
8
df.apply(lambda col: col.max() - col.min())
9

10
# ⚠️ 注意：apply 是 Python 循环，速度远慢于向量化操作
11
# 能用向量化的就不要用 apply：
12
# 不好：df.apply(lambda row: row['salary'] * 1.1, axis=1)
13
# 好：  df['salary'] * 1.1

1
import pandas as pd
2

3
chunk_size = 100_000  # 每次读取 10 万行
4

5
# 方式一：显式迭代
6
results = []
7
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size):
8
    # 对每个 chunk 进行处理
9
    filtered = chunk[chunk['status'] == 'active']
10
    results.append(filtered)
11

12
# 合并结果
13
df_result = pd.concat(results, ignore_index=True)

1
import pandas as pd
2

3
chunk_size = 200_000
4
agg_result = {}
5

6
for chunk in pd.read_csv('orders.csv', chunksize=chunk_size,
7
                          usecols=['order_type', 'amount'],  # 只读取需要的列
8
                          dtype={'amount': 'float32'}):      # 优化 dtype
9
    # 按 order_type 分组求和
10
    group_sum = chunk.groupby('order_type')['amount'].sum()
11
    for key, val in group_sum.items():
12
        agg_result[key] = agg_result.get(key, 0) + val
13

14
# 转为 DataFrame
15
result = pd.Series(agg_result).sort_values(ascending=False)
16
print(result)

1
# 只读取需要的列（usecols），指定 dtype，大幅提升速度和减少内存
2
df = pd.read_csv(
3
    'large_data.csv',
4
    usecols=['user_id', 'event_type', 'timestamp', 'amount'],  # 只读4列
5
    dtype={
6
        'user_id':    'int32',    # 默认 int64，节省一半
7
        'event_type': 'category', # 字符串 → category，节省 90%
8
        'amount':     'float32',  # 默认 float64，节省一半
9
    },
10
    parse_dates=['timestamp'],    # 直接解析为 datetime
11
    nrows=None,                   # None = 读全部；设数字 = 只读前 N 行（用于抽样调试）
12
)

1
import pandas as pd
2
import numpy as np
3

4
def optimize_dtypes(df: pd.DataFrame) -> pd.DataFrame:
5
    """自动优化 DataFrame 的数据类型，减少内存占用"""
6

7
    for col in df.columns:
8
        col_type = df[col].dtype
9

10
        # 整数列：降级到最小可用整数类型
11
        if col_type in ['int64', 'int32', 'int16']:
12
            col_min = df[col].min()
13
            col_max = df[col].max()
14
            if col_min >= np.iinfo(np.int8).min and col_max <= np.iinfo(np.int8).max:
15
                df[col] = df[col].astype(np.int8)
16
            elif col_min >= np.iinfo(np.int16).min and col_max <= np.iinfo(np.int16).max:
17
                df[col] = df[col].astype(np.int16)
18
            elif col_min >= np.iinfo(np.int32).min and col_max <= np.iinfo(np.int32).max:
19
                df[col] = df[col].astype(np.int32)
20

21
        # 浮点列：float64 → float32（精度略降，但通常够用）
22
        elif col_type == 'float64':
23
            df[col] = df[col].astype(np.float32)
24

25
        # 字符串/object 列：低基数的转 category
26
        elif col_type == 'object':
27
            num_unique = df[col].nunique()
28
            if num_unique / len(df) < 0.5:  # 唯一值少于 50%，转 category
29
                df[col] = df[col].astype('category')
30

31
    return df
32

33
# 使用示例
34
print(f"优化前内存：{df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")
35
df = optimize_dtypes(df)
36
print(f"优化后内存：{df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")

1
# 对比：object vs category
2
import pandas as pd
3
import numpy as np
4

5
# 模拟 100 万行，只有 5 种城市
6
n = 1_000_000
7
cities = ['北京', '上海', '广州', '深圳', '成都']
8
df = pd.DataFrame({'city': np.random.choice(cities, n)})
9

10
# object 类型（默认）
11
mem_object = df['city'].memory_usage(deep=True) / 1024 / 1024
12
print(f"object  类型内存：{mem_object:.1f} MB")   # 约 62 MB
13

14
# category 类型
15
df['city_cat'] = df['city'].astype('category')
16
mem_cat = df['city_cat'].memory_usage(deep=True) / 1024 / 1024
17
print(f"category 类型内存：{mem_cat:.1f} MB")     # 约 1 MB（节省 98%！）
18

19
# category 类型在 groupby 时也更快
20
%timeit df.groupby('city')['city'].count()
21
%timeit df.groupby('city_cat')['city'].count()
22
# category groupby 通常快 2-5x

1
# 测试：读取 CSV + groupby + filter 整体流程
2

3
# Pandas（单线程）
4
import pandas as pd
5
import time
6

7
t0 = time.perf_counter()
8
df = pd.read_csv('data_1M.csv')
9
result = df[df['status'] == 'active'].groupby('city')['amount'].sum()
10
print(f"Pandas: {time.perf_counter() - t0:.3f}s")   # 约 1.2s
11

12
# Polars（多线程 + 惰性求值）
13
import polars as pl
14

15
t0 = time.perf_counter()
16
result = (
17
    pl.scan_csv('data_1M.csv')        # scan = 惰性，不立即读取
18
    .filter(pl.col('status') == 'active')
19
    .group_by('city')
20
    .agg(pl.col('amount').sum())
21
    .collect()                          # collect = 触发实际计算
22
)
23
print(f"Polars: {time.perf_counter() - t0:.3f}s")   # 约 0.15s（快约 8x）

1
# Pandas ↔ Polars 互转
2
import pandas as pd
3
import polars as pl
4

5
pandas_df = pd.DataFrame({'a': [1, 2, 3]})
6

7
# Pandas → Polars
8
polars_df = pl.from_pandas(pandas_df)
9

10
# Polars → Pandas
11
pandas_df2 = polars_df.to_pandas()