企业官网建设流程全解析

23. applymap 与管道

1. 概述

applymap()是 DataFrame 的元素级函数应用方法，可以对每个单元格应用函数。管道（pipe()）则是一种函数链式调用方式，可以将多个数据处理步骤串联起来，使代码更清晰、更易维护。

importpandasaspdimportnumpyasnp# 创建示例数据df=pd.DataFrame({'A':[1,2,3,4,5],'B':[10,20,30,40,50],'C':[100,200,300,400,500],'D':['a','b','c','d','e']})print("原始数据:")print(df)

2. applymap() 方法

2.1 基本用法

applymap()对 DataFrame 的每个元素应用函数。

# 将所有元素乘以2print("每个元素乘以2:")print(df[['A','B','C']].applymap(lambdax:x*2))# 格式化所有数值df_formatted=df[['A','B','C']].applymap(lambdax:f"{x:.2f}")print("\n格式化:")print(df_formatted)

2.2 数据类型转换

# 将所有数值转为字符串df_str=df[['A','B','C']].applymap(str)print("转为字符串:")print(df_str.dtypes)# 添加前缀df_prefixed=df[['A','B','C']].applymap(lambdax:f"值_{x}")print("\n添加前缀:")print(df_prefixed)

2.3 条件格式化

# 根据值的大小进行格式化defformat_value(x):ifisinstance(x,(int,float)):ifx>30:returnf"**{x}**"elifx>10:returnf"*{x}*"returnstr(x)df_formatted=df.applymap(format_value)print("条件格式化:")print(df_formatted)

2.4 数据清洗

# 创建包含空格的示例数据df_dirty=pd.DataFrame({'姓名':[' 张三 ','李四',' 王五 '],'年龄':['25','30','28'],'城市':[' 北京','上海 ',' 广州 ']})print("脏数据:")print(df_dirty)# 批量清洗defclean_data(x):ifisinstance(x,str):returnx.strip()# 去除空格returnx df_clean=df_dirty.applymap(clean_data)print("\n清洗后:")print(df_clean)

3. pipe() 管道方法

3.1 基本用法

pipe()允许将 DataFrame 传递给函数，实现链式操作。

# 定义数据处理函数defremove_outliers(df,column,threshold=3):"""移除异常值"""mean=df[column].mean()std=df[column].std()returndf[abs(df[column]-mean)<=threshold*std]defscale_column(df,column):"""标准化列"""df[column]=(df[column]-df[column].mean())/df[column].std()returndfdefadd_derived_columns(df):"""添加派生列"""df['A_squared']=df['A']**2df['B_squared']=df['B']**2returndf# 使用 pipe 链式调用df_result=(df.pipe(remove_outliers,'A').pipe(scale_column,'B').pipe(add_derived_columns))print("管道处理结果:")print(df_result)

3.2 多个参数传递

# 带多个参数的函数defprocess_data(df,columns,multiplier=2,add_constant=0):forcolincolumns:df[col]=df[col]*multiplier+add_constantreturndf# 使用 pipe 传递参数df_processed=df.pipe(process_data,['A','B','C'],multiplier=3,add_constant=10)print("带参数处理:")print(df_processed)

3.3 管道与 applymap 结合

# 数据标准化处理流程defstandardize_numeric(df):"""标准化数值列"""numeric_cols=df.select_dtypes(include=['int64','float64']).columnsforcolinnumeric_cols:df[col]=(df[col]-df[col].mean())/df[col].std()returndfdefformat_output(df):"""格式化输出"""numeric_cols=df.select_dtypes(include=['int64','float64']).columns df[numeric_cols]=df[numeric_cols].applymap(lambdax:f"{x:.2f}")returndfdefadd_summary(df):"""添加汇总行"""df.loc['汇总']=df.select_dtypes(include=['object']).apply(lambdax:'—')returndf# 管道链df_result=(df.pipe(standardize_numeric).pipe(format_output).pipe(add_summary))print("完整处理流程:")print(df_result)

4. 链式操作对比

4.1 传统方式 vs 管道方式

# 传统方式（嵌套）df_temp=remove_outliers(df,'A')df_temp=scale_column(df_temp,'B')df_temp=add_derived_columns(df_temp)# 管道方式（链式）df_result=(df.pipe(remove_outliers,'A').pipe(scale_column,'B').pipe(add_derived_columns))print("两种方式结果相同:",df_temp.equals(df_result))

4.2 方法链 vs 管道

# 方法链（内置方法）df_chain=(df.dropna().sort_values('A').reset_index(drop=True))# 管道（自定义函数）df_pipe=(df.pipe(lambdax:x.dropna()).pipe(lambdax:x.sort_values('A')).pipe(lambdax:x.reset_index(drop=True)))

5. 完整示例：数据清洗管道

# 创建脏数据np.random.seed(42)dirty_data=pd.DataFrame({'id':[1,2,3,4,5,6,7,8,9,10],'name':[' 张三','李四 ',' 王五 ','赵六',None,'钱七',' 孙八','周九','吴十','郑十一'],'age':[25,-5,30,150,28,32,0,35,130,27],'salary':[5000,8000,-1000,12000,10000,None,15000,9000,25000,11000],'dept':['技术','销售','技术','市场','销售','技术',None,'市场','销售','技术']})print("="*60)print("数据清洗管道")print("="*60)print("\n原始脏数据:")print(dirty_data)# 定义清洗函数defclean_names(df):"""清洗姓名列"""if'name'indf.columns:df['name']=df['name'].fillna('未知')df['name']=df['name'].str.strip()returndfdefclean_ages(df):"""清洗年龄列（限制在0-120之间）"""if'age'indf.columns:df['age']=df['age'].clip(0,120)df['age']=df['age'].fillna(df['age'].median())returndfdefclean_salaries(df):"""清洗工资列"""if'salary'indf.columns:df['salary']=df['salary'].clip(0,50000)df['salary']=df['salary'].fillna(df['salary'].median())returndfdefclean_departments(df):"""清洗部门列"""if'dept'indf.columns:df['dept']=df['dept'].fillna('未知')df['dept']=df['dept'].str.strip()returndfdefadd_derived_columns_pipe(df):"""添加派生列"""df['age_group']=pd.cut(df['age'],bins=[0,30,60,120],labels=['青年','中年','老年'])df['salary_level']=pd.cut(df['salary'],bins=[0,5000,10000,20000,50000],labels=['低','中','高','超高'])returndfdefremove_invalid_rows(df):"""删除无效行"""df=df.dropna(subset=['id','name'])returndf# 使用管道执行所有清洗步骤cleaned_data=(dirty_data.pipe(clean_names).pipe(clean_ages).pipe(clean_salaries).pipe(clean_departments).pipe(remove_invalid_rows).pipe(add_derived_columns_pipe))print("\n清洗后数据:")print(cleaned_data)print("\n清洗前后对比:")print(f"清洗前行数:{len(dirty_data)}")print(f"清洗后行数:{len(cleaned_data)}")

6. applymap vs apply vs pipe 对比

7. 性能考虑

# applymap 性能importtime large_df=pd.DataFrame(np.random.randn(1000,100))start=time.time()large_df.applymap(lambdax:x*2)print(f"applymap 耗时:{time.time()-start:.4f}秒")# 向量化更快start=time.time()large_df*2print(f"向量化耗时:{time.time()-start:.4f}秒")

8. 总结