本文目录导读：
Python + Pandas (最常用)
基础汇总脚本
import pandas as pd
import numpy as np
# 读取数据
df = pd.read_csv('data.csv')
# 基本汇总
summary = {
'总行数': len(df),
'总销售额': df['销售额'].sum(),
'平均销售额': df['销售额'].mean(),
'最大销售额': df['销售额'].max(),
'最小销售额': df['销售额'].min()
}
# 分组汇总
grouped = df.groupby('部门').agg({
'销售额': ['sum', 'mean', 'count'],
'成本': 'sum',
'利润': ['sum', 'mean']
})
# 导出结果
summary_df = pd.DataFrame([summary])
summary_df.to_excel('汇总结果.xlsx', index=False)
高级汇总脚本
import pandas as pd
from datetime import datetime
class DataSummarizer:
def __init__(self, file_path):
self.df = pd.read_csv(file_path)
def basic_stats(self, columns):
"""基础统计"""
return self.df[columns].describe()
def group_summary(self, group_col, agg_cols):
"""分组汇总"""
return self.df.groupby(group_col)[agg_cols].sum()
def time_series_summary(self, date_col, value_col, freq='M'):
"""时间序列汇总"""
self.df[date_col] = pd.to_datetime(self.df[date_col])
return self.df.set_index(date_col).resample(freq)[value_col].sum()
def cross_tab(self, row_col, col_col, value_col):
"""交叉表汇总"""
return pd.pivot_table(
self.df,
values=value_col,
index=row_col,
columns=col_col,
aggfunc=np.sum
)
# 使用示例
summarizer = DataSummarizer('销售数据.csv')
monthly_sales = summarizer.time_series_summary('日期', '销售额', 'M')
SQL 脚本
基本汇总查询
-- 基本统计
SELECT
COUNT(*) as 总记录数,
SUM(销售额) as 总销售额,
AVG(销售额) as 平均销售额,
MAX(销售额) as 最大销售额,
MIN(销售额) as 最小销售额
FROM 销售表;
-- 分组汇总
SELECT
部门,
SUM(销售额) as 部门总销售额,
AVG(销售额) as 部门平均销售额,
COUNT(*) as 部门订单数
FROM 销售表
GROUP BY 部门
HAVING SUM(销售额) > 10000
ORDER BY 部门总销售额 DESC;
-- 多层汇总
SELECT
COALESCE(部门, '总计') as 部门,
COALESCE(产品类别, '小计') as 产品类别,
SUM(销售额) as 销售额合计
FROM 销售表
GROUP BY ROLLUP(部门, 产品类别);
存储过程
CREATE PROCEDURE GenerateSalesSummary
@StartDate DATE,
@EndDate DATE
AS
BEGIN
-- 创建临时表存储汇总结果
CREATE TABLE #SalesSummary (
Category VARCHAR(50),
TotalSales DECIMAL(18,2),
OrderCount INT,
AvgOrderValue DECIMAL(18,2)
);
-- 插入汇总数据
INSERT INTO #SalesSummary
SELECT
ProductCategory,
SUM(SalesAmount),
COUNT(DISTINCT OrderID),
AVG(SalesAmount)
FROM Sales
WHERE OrderDate BETWEEN @StartDate AND @EndDate
GROUP BY ProductCategory;
-- 返回结果
SELECT * FROM #SalesSummary
ORDER BY TotalSales DESC;
-- 清理临时表
DROP TABLE #SalesSummary;
END;
Excel VBA 宏
Sub GenerateSummary()
Dim ws As Worksheet
Dim summaryWS As Worksheet
Dim lastRow As Long
Dim dataRange As Range
Dim dict As Object
' 设置数据工作表
Set ws = ThisWorkbook.Sheets("数据")
Set summaryWS = ThisWorkbook.Sheets("汇总")
' 获取数据范围
lastRow = ws.Cells(ws.Rows.Count, "A").End(xlUp).Row
Set dataRange = ws.Range("A1:D" & lastRow)
' 创建字典用于分组汇总
Set dict = CreateObject("Scripting.Dictionary")
' 遍历数据
For i = 2 To lastRow
category = ws.Cells(i, 1).Value
amount = ws.Cells(i, 3).Value
If dict.exists(category) Then
dict(category) = dict(category) + amount
Else
dict.Add category, amount
End If
Next i
' 输出汇总结果
summaryWS.Cells(1, 1) = "类别"
summaryWS.Cells(1, 2) = "汇总金额"
rowNum = 2
For Each key In dict.keys
summaryWS.Cells(rowNum, 1) = key
summaryWS.Cells(rowNum, 2) = dict(key)
rowNum = rowNum + 1
Next key
' 添加总计
summaryWS.Cells(rowNum, 1) = "总计"
summaryWS.Cells(rowNum, 2) = Application.WorksheetFunction.Sum(summaryWS.Range("B2:B" & rowNum - 1))
End Sub
Shell 脚本 (Linux)
#!/bin/bash
# CSV文件汇总脚本
FILE="sales_data.csv"
# 计算总销售额
total_sales=$(awk -F',' 'NR>1 {sum+=$3} END {print sum}' $FILE)
# 按部门汇总
echo "部门汇总报告"
echo "============"
awk -F',' 'NR>1 {
dept[$1] += $3
count[$1]++
} END {
for (d in dept) {
print d ": 总销售额=" dept[d] ", 订单数=" count[d]
}
}' $FILE
# 按月份汇总
echo ""
echo "月度汇总报告"
echo "============"
awk -F',' 'NR>1 {
split($2, date, "-")
month = date[1] "-" date[2]
monthly[month] += $3
} END {
for (m in monthly) {
print m ": " monthly[m]
}
}' $FILE
性能优化建议
大数据集处理
import pandas as pd
import dask.dataframe as dd
# 使用Dask处理大数据
ddf = dd.read_csv('large_data.csv')
# 分区计算
summary = ddf.groupby('category').agg({
'amount': ['sum', 'mean', 'count']
}).compute()
# 并行处理
from multiprocessing import Pool
def process_chunk(chunk_df):
return chunk_df.groupby('category')['amount'].sum()
# 分块处理
chunks = pd.read_csv('large_data.csv', chunksize=10000)
with Pool(4) as pool:
results = pool.map(process_chunk, chunks)
# 合并结果
final_result = pd.concat(results).groupby(level=0).sum()
内存优化
# 使用内存优化读取
dtypes = {
'category': 'category',
'amount': 'float32',
'date': 'datetime64[ns]'
}
df = pd.read_csv('data.csv', dtype=dtypes)
# 增量汇总
def incremental_summary(file_path):
summary = {}
for chunk in pd.read_csv(file_path, chunksize=5000):
for _, row in chunk.iterrows():
cat = row['category']
amount = row['amount']
if cat in summary:
summary[cat] += amount
else:
summary[cat] = amount
return summary
模板化脚本
# config.yaml
datasource:
type: csv
path: data.csv
delimiter: ','
columns:
group_by: category
value_cols: [amount, quantity]
aggregations:
- type: sum
columns: [amount, quantity]
- type: mean
columns: [amount]
- type: count
columns: [*]
output:
format: excel
path: summary_results.xlsx
# 通用汇总引擎
class SummaryEngine:
def __init__(self, config):
self.config = config
def run(self):
df = self.load_data()
result = self.aggregate(df)
self.save_result(result)
def aggregate(self, df):
group_col = self.config['columns']['group_by']
agg_dict = {}
for agg in self.config['aggregations']:
for col in agg['columns']:
if col == '*':
col = df.select_dtypes(include=[np.number]).columns
agg_dict[col] = agg['type']
return df.groupby(group_col).agg(agg_dict)
选择哪种方法取决于：
- 数据规模和复杂度
- 是否需要自动化运行
- 团队技术栈
- 实时性要求
建议从简单的 Python + Pandas 开始,逐步扩展功能。
