Python Pandas 高效入门指南：从零基础到实战数据处理技巧

引言：为什么选择Pandas？

Pandas是Python中最流行的数据处理库，它提供了高性能、易用的数据结构和数据分析工具。无论你是数据分析师、数据科学家还是软件工程师，掌握Pandas都能让你在数据处理方面事半功倍。

学习目标

理解Pandas的核心数据结构
掌握数据读取与写入操作
熟练使用数据筛选、清洗和转换技巧
学会分组聚合与时间序列处理
通过实战案例巩固所学知识

第一部分：Pandas基础概念

1.1 核心数据结构

Series（序列）

Series是一维标记数组，可以存储任意数据类型。

import pandas as pd import numpy as np # 创建Series s = pd.Series([1, 3, 5, np.nan, 6, 8]) print(s)

输出：

0 1.0 1 3.0 2 5.0 3 NaN 4 6.0 5 8.0 dtype: float64

DataFrame（数据框）

DataFrame是二维表格型数据结构，是最常用的Pandas对象。

# 创建DataFrame data = { 'Name': ['Alice', 'Bob', 'Charlie'], 'Age': [25, 30, 35], 'City': ['New York', 'London', 'Paris'] } df = pd.DataFrame(data) print(df)

输出：

 Name Age City 0 Alice 25 New York 1 Bob 30 London 2 Charlie 35 Paris

1.2 基本属性和方法

# 查看DataFrame基本信息 print("数据形状:", df.shape) # (3, 3) print("数据类型:n", df.dtypes) print("列名:", df.columns.tolist()) print("前3行数据:n", df.head(3)) print("统计摘要:n", df.describe())

第二部分：数据读取与写入

2.1 读取数据

Pandas支持多种文件格式，包括CSV、Excel、JSON、SQL等。

读取CSV文件

# 基本读取 df = pd.read_csv('data.csv') # 处理中文编码 df = pd.read_csv('data.csv', encoding='utf-8') # 指定列索引 df = pd.read_csv('data.csv', index_col='ID') # 处理大文件（分块读取） chunk_size = 10000 for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size): process(chunk)

读取Excel文件

# 读取Excel df = pd.read_excel('data.xlsx', sheet_name='Sheet1') # 读取多个sheet excel_file = pd.ExcelFile('data.xlsx') sheets = {sheet: pd.read_excel(excel_file, sheet_name=sheet) for sheet in excel_file.sheet_names}

读取JSON数据

# 读取JSON df = pd.read_json('data.json') # 从API响应读取 import requests response = requests.get('https://api.example.com/data') df = pd.DataFrame(response.json())

2.2 写入数据

# 写入CSV（不包含索引） df.to_csv('output.csv', index=False) # 写入Excel（指定格式） df.to_excel('output.xlsx', sheet_name='Data', index=False) # 写入JSON df.to_json('output.json', orient='records', indent=2)

第三部分：数据查看与选择

3.1 查看数据

# 创建示例数据 np.random.seed(42) df = pd.DataFrame({ 'Name': ['Alice', 'Bob', 'Charlie', 'David', 'Eva'], 'Age': [25, 30, 35, 28, 32], 'Salary': [50000, 60000, 70000, 55000, 65000], 'Department': ['HR', 'IT', 'Finance', 'IT', 'HR'], 'JoinDate': pd.date_range('2020-01-01', periods=5, freq='M') }) # 基本查看 print("前5行:n", df.head()) print("后3行:n", df.tail(3)) print("随机抽样3行:n", df.sample(3)) print("数据信息:") df.info()

3.2 数据选择

选择列

# 单列选择 names = df['Name'] # 返回Series names = df.Name # 同上（属性方式） # 多列选择 subset = df[['Name', 'Salary']] # 返回DataFrame

选择行（索引器）

# loc：基于标签的索引 print(df.loc[0]) # 第0行 print(df.loc[1:3]) # 第1到3行（包含3） print(df.loc[1:3, ['Name', 'Age']]) # 行列同时选择 # iloc：基于位置的索引 print(df.iloc[0]) # 第0行 print(df.iloc[1:3]) # 第1到3行（不包含3） print(df.iloc[1:3, [0, 1]] # 行列同时选择 # 条件选择 print(df[df['Age'] > 28]) # 年龄大于28的行 print(df.query('Age > 28 & Department == "IT"')) # 使用query方法

3.3 高级索引技巧

# 设置索引 df_indexed = df.set_index('Name') # 重置索引 df_reset = df_indexed.reset_index() # 多级索引 df_multi = df.set_index(['Department', 'Name']) print(df_multi.loc['IT']) # 选择Department为IT的所有行

第四部分：数据清洗与预处理

4.1 处理缺失值

# 创建包含缺失值的数据 df_missing = pd.DataFrame({ 'A': [1, 2, np.nan, 4], 'B': [5, np.nan, 3, 8], 'C': [np.nan, 2, 3, 4] }) # 检查缺失值 print("缺失值统计:n", df_missing.isnull().sum()) print("缺失值位置:n", df_missing.isnull()) # 删除缺失值 df_dropped = df_missing.dropna() # 删除任何包含NaN的行 df_dropped_cols = df_missing.dropna(axis=1) # 删除包含NaN的列 df_dropped_thresh = df_missing.dropna(thresh=2) # 保留至少2个非空值的行 # 填充缺失值 df_filled = df_missing.fillna(0) # 用0填充 df_filled_mean = df_missing.fillna(df_missing.mean()) # 用均值填充 df_filled_forward = df_missing.fillna(method='ffill') # 前向填充 df_filled_backward = df_missing.fillna(method='bfill') # 后向填充 # 插值填充 df_interpolated = df_missing.interpolate() # 线性插值

4.2 数据类型转换

# 创建示例数据 df = pd.DataFrame({ 'Date': ['2023-01-01', '2023-01-02', '2023-01-03'], 'Price': ['100.5', '101.2', '102.8'], 'Quantity': ['10', '20', '30'], 'Category': ['A', 'B', 'A'] }) # 转换为datetime df['Date'] = pd.to_datetime(df['Date']) # 转换为数值类型 df['Price'] = df['Price'].astype(float) df['Quantity'] = df['排序

4.3 处理重复值

# 创建包含重复值的数据 df_duplicates = pd.DataFrame({ 'A': [1, 2, 2, 4], 'B': [5, 6, 6, 8], 'C': [7, 8, 8, 10] }) # 检查重复值 print("重复值数量:", df_duplicates.duplicated().sum()) # 删除重复值 df_unique = df_duplicates.drop_duplicates() df_unique_keep_last = df_duplicates.drop_duplicates(keep='last') df_unique_subset = df_duplicates.drop_duplicates(subset=['A', 'B'])

4.4 数据标准化与规范化

# 数据标准化（Z-score） from sklearn.preprocessing import StandardScaler scaler = StandardScaler() df[['Age', 'Salary']] = scaler.fit_transform(df[['Age', 'Salary']]) # 数据规范化（Min-Max） from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler() df[['Age', 'Salary']] = scaler.fit_transform(df[['Age', '2.5 数据类型转换

4.5 处理异常值

# 使用IQR方法检测异常值 def detect_outliers_iqr(df, column): Q1 = df[column].quantile(0.25) Q3 = df[column].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.2 * IQR return df[(df[column] < lower_bound) | (df[column] > upper_bound)] # 使用Z-score方法 def detect_outliers_zscore(df, column, threshold=3): z_scores = np.abs((df[column] - df[column].mean()) / df[column].std()) return df[z_scores > threshold] # 处理异常值（替换为边界值） def handle_outliers(df, column): Q1 = df[column].quantile(0.25) Q3 = df[column].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.2 * IQR df[column] = df[column].clip(lower=lower_bound, upper=upper_bound) return df

第五部分：数据转换与操作

5.1 数据排序

# 单列排序 df_sorted = df.sort_values('Salary', ascending=False) # 多列排序 df_sorted = df.sort_values(['Department', 'Salary'], ascending=[True, False]) # 按索引排序 df_sorted = df.sort_index()

5.2 数据分组与聚合

# 基本分组聚合 grouped = df.groupby('Department') print("部门平均工资:", grouped['Salary'].mean()) print("部门人数:", grouped['Name'].count()) # 多列聚合 agg_funcs = { 'Salary': ['mean', 'max', 'min', 'std'], 'Age': ['mean', 'median'] } department_stats = df.groupby('Department').agg(agg_funcs) print(department_stats) # 自定义聚合函数 def salary_range(series): return series.max() - series.min() custom_agg = df.groupby('Department').agg({ 'Salary': [salary_range, 'mean'], 'Age': 'max' }) print(custom_agg) # transform和apply df['Dept_Avg_Salary'] = df.groupby('Department')['Salary'].transform('mean') df['Salary_Rank'] = df.groupby('Department')['Salary'].rank(ascending=False)

5.3 数据合并与连接

# 创建示例数据 df1 = pd.DataFrame({ 'ID': [1, 2, 3], 'Name': ['Alice', 'Bob', 'Charlie'], 'Age': [25, 30, 35] }) df2 = pd.DataFrame({ 'ID': [2, 3, 4], 'City': ['London', 'Paris', 'Tokyo'], 'Salary': [60000, 70000, 80000] }) # merge操作 merged_inner = pd.merge(df1, df2, on='ID', how='inner') # 内连接 merged_left = pd.merge(df1, df2, on='ID', how='left') # 左连接 merged_right = pd.merge(df1, df2, on='ID', how='right') # 右连接 merged_outer = pd.merge(df1, df2, on='ID', how='outer') # 外连接 # concat操作 concat_rows = pd.concat([df1, df2], axis=0) # 纵向合并 concat_cols = pd.concat([df1, df2], axis=1) # 横向合并 # join操作 df1.set_index('ID', inplace=True) df2.set_index('ID', inplace=True) joined = df1.join(df2, how='inner')

5.4 数据重塑与透视

# 创建示例数据 df_pivot = pd.DataFrame({ 'Date': ['2023-01-01', '2023-01-01', '2023-01-02', '2023-01-02'], 'Product': ['A', 'B', 'A', 'B'], 'Sales': [100, 150, 120, 180], 'Profit': [20, 30, 25, 35] }) # 透视表 pivot_table = pd.pivot_table( df_pivot, values=['Sales', 'Profit'], index='Date', columns='Product', aggfunc='sum' ) # melt操作（宽变长） melted = pd.melt( df_pivot, id_vars=['Date', 'Product'], value_vars=['Sales', 'Profit'], var_name='Metric', value_name='Value' ) # stack和unstack stacked = df_pivot.set_index(['Date', 'Product']).stack() unstacked = stacked.unstack()

5.5 字符串操作

# 创建示例数据 df_text = pd.DataFrame({ 'Name': ['Alice Smith', 'Bob Johnson', 'Charlie Brown'], 'Email': ['alice@example.com', 'bob@example.com', 'charlie@example.com'], 'Company': ['ABC Inc.', 'XYZ Corp.', 'DEF Ltd.'] }) # 字符串方法 df_text['First_Name'] = df_text['Name'].str.split().str[0] df_text['Domain'] = df_text['Email'].str.split('@').str[1] df_text['Company_Clean'] = df_text['Company'].str.replace(r'.', '', regex=True) df_text['Name_Lower'] = df_text['Name'].str.lower() df_text['Name_Upper'] =df_text['Name'].str.upper() df_text['Name_Title'] = df_text['Name'].str.title() # 正则表达式 df_text['Has_Gmail'] = df_text['Email'].str.contains(r'@gmail.com') df_text['Company_Abbrev'] = df_text['Company'].str.extract(r'([A-Z]{2,})')

5.6 时间序列操作

# 创建时间序列数据 dates = pd.date_range('2023-01-01', periods=10, freq='D') ts = pd.Series(np.random.randn(10), index=dates) # 时间提取 df_time = pd.DataFrame({'date': dates, 'value': np.random.randn(10)}) df_time['year'] = df_time['date'].dt.year df_time['month'] = df_time['date'].dt.month df_time['day'] = df_time['date'].dt.day df_time['weekday'] = df_time['date'].dt.weekday df_time['is_month_end'] = df_time['date'].dt.is_month_end # 重采样 daily_resample = df_time.set_index('date').resample('D').mean() weekly_resample = df_time.set_index('2.5 数据类型转换

第六部分：实战案例：销售数据分析

6.1 案例背景

假设我们是一家电商公司的数据分析师，需要分析2023年的销售数据。

6.2 数据准备

import pandas as pd import numpy as np import matplotlib.pyplot as plt # 创建模拟销售数据 np.random.seed(42) n = 1000 sales_data = pd.DataFrame({ 'OrderID': range(1001, 1001 + n), 'Date': pd.date_range('2023-01-01', periods=n, freq='H'), 'CustomerID': np.random.randint(100, 200, n), 'ProductID': np.random.randint(1, 10, n), 'Category': np.random.choice(['Electronics', 'Clothing', 'Books', 'Home'], n), 'Quantity': np.random.randint(1, 10, n), 'UnitPrice': np.random.uniform(10, 200, n).round(2), 'Discount': np.random.choice([0, 0.1, 0.2, 0.3], n, p=[0.7, 0.15, 0.1, 0.05]) }) # 计算总销售额 sales_data['TotalAmount'] = sales_data['Quantity'] * sales_data['UnitPrice'] * (1 - sales_data['Discount']) # 添加一些缺失值和异常值 sales_data.loc[0:5, 'Discount'] = np.nan sales_data.loc[10:15, 'UnitPrice'] = 1000 # 异常高价 sales_data.loc[20:25, 'Quantity'] = 0 # 异常数量 print("数据预览:") print(sales_data.head()) print("n数据信息:") sales_data.info()

6.3 数据清洗

# 1. 处理缺失值 print("缺失值统计:n", sales_data.isnull().sum()) sales_data['Discount'].fillna(0, inplace=True) # 2. 处理异常值 # 修正异常单价（超过3倍标准差） price_mean = sales_data['UnitPrice'].mean() price_std = sales_data['UnitPrice'].std() price_upper = price_mean + 3 * price_std sales_data.loc[sales_data['UnitPrice'] > price_upper, 'UnitPrice'] = price_upper # 修正异常数量（小于等于0的设为1） sales_data.loc[sales_data['Quantity'] <= 0, 'Quantity'] = 1 # 3. 重新计算总金额 sales_data['TotalAmount'] = sales_data['Quantity'] * sales_data['UnitPrice'] * (1 - sales_data['Discount']) # 4. 检查清洗结果 print("n清洗后数据统计:") print(sales_data.describe())

6.4 数据分析

# 1. 基础统计 print("总销售额:", sales_data['TotalAmount'].sum()) print("平均订单金额:", sales_data['TotalAmount'].mean()) print("订单总数:", len(sales_data)) print("客户数量:", sales_data['CustomerID'].nunique()) # 2. 按类别分析 category_analysis = sales_data.groupby('Category').agg({ 'TotalAmount': ['sum', 'mean', 'count'], 'Quantity': 'sum' }).round(2) print("n按类别分析:n", category_analysis) # 3. 按月份分析 sales_data['Month'] = sales_data['Date'].dt.month monthly_sales = sales_data.groupby('Month')['TotalAmount'].sum() print("n月度销售额:n", monthly_sales) # 4. 热门产品分析 top_products = sales_data.groupby('ProductID')['TotalAmount'].sum().nlargest(5) print("n销售额前5的产品:n", top_products) # 5. 客户价值分析 customer_value = sales_data.groupby('CustomerID').agg({ 'TotalAmount': ['sum', 'mean', 'count'] }).round(2) customer_value.columns = ['Total_Spend', 'Avg_Order_Value', 'Order_Count'] customer_value = customer_value.sort_values('Total_Spend', ascending=False) print("n客户价值排名前5:n", customer_value.head())

6.5 数据可视化

# 1. 月度销售趋势 plt.figure(figsize=(12, 6)) monthly_sales.plot(kind='bar', color='skyblue') plt.title('2023 Monthly Sales Trend') plt.xlabel('Month') plt.ylabel('Total Sales') plt.xticks(rotation=0) plt.tight_layout() plt.show() # 2. 类别销售占比 plt.figure(figsize=(8, 8)) category_sales = sales_data.groupby('Category')['TotalAmount'].sum() plt.pie(category_sales.values, labels=category_sales.index, autopct='%1.1f%%') plt.title('Sales by Category') plt.show() # 3. 销售额分布 plt.figure(figsize=(10, 6)) sales_data['TotalAmount'].hist(bins=50, color='lightgreen', alpha=0.7) plt.title('Distribution of Order Amounts') plt.xlabel('Order Amount') plt.ylabel('Frequency') plt.show()

6.6 高级分析

# 1. RFM分析（最近购买时间、购买频率、消费金额） snapshot_date = sales_data['Date'].max() + pd.Timedelta(days=1) rfm = sales_data.groupby('CustomerID').agg({ 'Date': lambda x: (snapshot_date - x.max()).days, # Recency 'OrderID': 'count', # Frequency 'TotalAmount': 'sum' # Monetary }) rfm.columns = ['Recency', 'Frequency', 'Monetary'] # 2. 客户分层（使用四分位数） rfm['R_Score'] = pd.qcut(rfm['Recency'], 4, labels=[4, 3, 2, 1]) # 越小越好 rfm['F_Score'] = pd.qcut(rfm['Frequency'].rank(method='first'), 4, labels=[1, 2, 3, 4]) rfm['M_Score'] = pd.qcut(rfm['Monetary'], 4, labels=[1, 2, 3, 4]) rfm['RFM_Segment'] = rfm['R_Score'].astype(str) + rfm['F_Score'].astype(str) + rfm['M_Score'].astype(str) rfm['RFM_Score'] = rfm['R_Score'].astype(int) + rfm['F_Score'].astype(int) + rfm['M_Score'].astype(int) # 3. 客户分类 def customer_segment(score): if score >= 10: return 'VIP' elif score >= 8: return 'Loyal' elif score >= 6: return 'Potential' else: return 'Need Attention' rfm['Segment'] = rfm['RFM_Score'].apply(customer_segment) print("n客户分层统计:n", rfm['Segment'].value_counts())

6.7 结果导出

# 导出清洗后的数据 sales_data.to_csv('cleaned_sales_data.csv', index=False) # 导出分析结果 category_analysis.to_csv('category_analysis.csv') monthly_sales.to_csv('monthly_sales.csv') customer_value.head(10).to_csv('top_customers.csv') rfm.to_csv('rfm_analysis.csv') print("n所有分析结果已导出！")

第七部分：性能优化技巧

7.1 使用向量化操作

# 避免使用循环 # 不好的做法 def calculate_total_bad(df): result = [] for i in range(len(df)): result.append(df.loc[i, 'Quantity'] * df.loc[i, 'UnitPrice']) return result # 好的做法（向量化） def calculate_total_good(df): return df['Quantity'] * df['UnitPrice'] # 性能对比 import time # 测试数据 test_df = pd.DataFrame({ 'Quantity': np.random.randint(1, 10, 100000), 'UnitPrice': np.random.uniform(10, 200, 100000) }) # 测试循环 start = time.time() result_bad = calculate_total_bad(test_df) time_bad = time.time() - start # 测试向量化 start = time.time() result_good = calculate_total_good(test_df) time_good = time.time() - start print(f"循环耗时: {time_bad:.4f}秒") print(f"向量化耗时: {time_good:.4f}秒") print(f"性能提升: {time_bad/time_good:.1f}倍")

7.2 使用适当的数据类型

# 优化内存使用 def optimize_memory(df): df_optimized = df.copy() # 优化整数类型 for col in df_optimized.select_dtypes(include=['int64']).columns: df_optimized[col] = pd.to_numeric(df_optimized[col], downcast='integer') # 优化浮点数类型 for col in df_optimized.select_dtypes(include=['float64']).columns: df_optimized[col] = pd.to_numeric(df_optimized[col], downcast='float') # 优化对象类型（分类） for col in df_optimized.select_dtypes(include=['object']).columns: if df_optimized[col].nunique() / len(df_optimized) < 0.5: df_optimized[col] = df_optimized[col].astype('category') return df_optimized # 使用示例 df_optimized = optimize_memory(sales_data) print("优化前后内存使用对比:") print("原始内存:", sales_data.memory_usage(deep=True).sum() / 1024**2, "MB") print("优化后内存:", df_optimized.memory_usage(deep=True).sum() / 0.001, "MB")

7.3 使用query和eval方法

# query方法用于复杂筛选 # 普通方法 result1 = df[(df['Age'] > 25) & (df['Salary'] < 70000) & (df['Department'] == 'IT')] # query方法 result2 = df.query('Age > 25 and Salary < 70000 and Department == "IT"') # eval方法用于高效计算 # 普通方法 df['Total'] = df['Quantity'] * df['UnitPrice'] * (1 - df['Discount']) # eval方法（对于大数据集更快） df.eval('Total = Quantity * UnitPrice * (1 - Discount)', inplace=True)

7.4 分块处理大数据

# 处理无法一次性读入内存的大文件 def process_large_file(file_path, chunk_size=10000): results = [] for chunk in pd.read_csv(file_path, chunksize=chunk_size): # 处理每个chunk chunk['Total'] = chunk['Quantity'] * chunk['UnitPrice'] chunk_summary = chunk.groupby('Category')['Total'].sum() results.append(chunk_summary) # 合并结果 final_result = pd.concat(results).groupby(level=0).sum() return final_result # 使用示例 # result = process_large_file('large_sales_data.csv')

7.5 使用Categorical类型

# 将重复字符串转换为分类 df['Department'] = df['Department'].astype('category') # 分类类型的优势 # 1. 内存占用更小 # 2. 操作速度更快 # 3. 支持排序和比较 # 查看分类信息 print(df['Department'].cat.categories) print(df['Department'].cat.codes)

第八部分：常见问题与解决方案

8.1 SettingWithCopyWarning警告

# 问题代码 df[df['Age'] > 30]['Salary'] = 70000 # 会触发警告 # 解决方案1：使用.loc df.loc[df['Age'] > 30, 'Salary'] = 70000 # 解决方案2：使用.copy() df_filtered = df[df['Age'] > 30].copy() df_filtered['Salary'] = 70000

8.2 内存不足问题

# 1. 读取时指定数据类型 df = pd.read_csv('large_file.csv', dtype={'Category': 'category'}) # 2. 只读取需要的列 df = pd.read_csv('large_file.csv', usecols=['Date', 'Category', 'Amount']) # 3. 分块读取 for chunk in pd.read_csv('large_file.csv', chunksize=10000): process(chunk)

8.3 合并时的索引问题

# 问题：合并后索引混乱 merged = pd.merge(df1, df2, on='ID', how='left') # 结果索引是df1的索引，可能不连续 # 解决方案：重置索引 merged = merged.reset_index(drop=True)

8.4 中文编码问题

# 读取中文CSV try: df = pd.read_csv('data.csv', encoding='utf-8') except UnicodeDecodeError: try: df = pd.read_csv('data.csv', encoding='gbk') except: df = pd.read_csv('data.csv', encoding='gb18030')

8.5 日期格式混乱

# 统一日期格式 df['Date'] = pd.to_datetime(df['Date'], errors='coerce') # 无法转换的变为NaT # 处理多种格式 date_formats = ['%Y-%m-%d', '%d/%m/%Y', '%m-%d-%Y'] for fmt in date_formats: mask = df['Date'].isna() if mask.any(): df.loc[mask, 'Date'] = pd.to_datetime(df.loc[mask, 'Date'], format=fmt, errors='ignore')

第九部分：进阶技巧与最佳实践

9.1 链式操作

# 使用链式操作保持代码简洁 result = (df .query('Age > 25') .assign(Salary_Tax=lambda x: x['Salary'] * 0.2) .groupby('Department') .agg({'Salary_Tax': 'sum'}) .sort_values('Salary_Tax', ascending=False) )

9.2 自定义函数应用

# 使用apply def calculate_bonus(row): if row['Age'] > 30 and row['Salary'] > 60000: return row['Salary'] * 0.1 else: return row['Salary'] * 0.05 df['Bonus'] = df.apply(calculate_bonus, axis=1) # 使用map进行列转换 salary_map = {'HR': 50000, 'IT': 60000, 'Finance': 70000} df['Base_Salary'] = df['Department'].map(salary_map)

9.3 并行处理

from multiprocessing import Pool def process_chunk(chunk): return chunk.groupby('Category')['Amount'].sum() def parallel_process(df, n_workers=4): df_split = np.array_split(df, n_workers) with Pool(n_workers) as p: results = p.map(process_chunk, df_split) return pd.concat(results).groupby(level=0).sum()

9.4 使用配置管理

# config.py CONFIG = { 'DATA_PATH': 'data/', 'OUTPUT_PATH': 'results/', 'CHUNK_SIZE': 10000, 'CATEGORIES': ['Electronics', 'Clothing', 'Books', 'Home'] } # 使用配置 import config df = pd.read_csv(f"{config.CONFIG['DATA_PATH']}sales.csv")

9.5 单元测试

import unittest class TestDataProcessing(unittest.TestCase): def setUp(self): self.df = pd.DataFrame({ 'A': [1, 2, 3], 'B': [4, 5, 6] }) def test_sum_calculation(self): result = self.df['A'] + self.df['B'] expected = pd.Series([5, 7, 9]) pd.testing.assert_series_equal(result, expected) def test_filter(self): result = self.df[self.df['A'] > 1] self.assertEqual(len(result), 2) # 运行测试 # unittest.main()

第十部分：学习资源与进阶路径

10.1 推荐资源

官方文档: pandas.pydata.org
书籍: 《Python for Data Analysis》by Wes McKinney
在线课程: Coursera, DataCamp
社区: Stack Overflow, GitHub

10.2 进阶学习路径

基础阶段: 掌握DataFrame操作、数据清洗
中级阶段: 学习分组聚合、时间序列、合并操作
高级阶段: 性能优化、自定义函数、并行处理
专业阶段: 与其他库集成（NumPy, Scikit-learn, Matplotlib）

10.3 常用库组合

# 数据科学生态 import pandas as pd # 数据处理 import numpy as np # 数值计算 import matplotlib.pyplot as plt # 可视化 import seaborn as sns # 高级可视化 from sklearn.model_selection import train_test_split # 机器学习 from sklearn.preprocessing import StandardScaler # 特征工程