怎样实现数据汇总计算脚本

本文目录导读：

1. Python + Pandas (最常用)
2. SQL 脚本
3. Excel VBA 宏
4. Shell 脚本 (Linux)
5. 性能优化建议
6. 模板化脚本

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 开始,逐步扩展功能。