怎样实现动态分组聚合脚本

本文目录导读：

1. Python 动态分组聚合
2. SQL 动态分组聚合
3. JavaScript 实现
4. 配置驱动的通用实现
5. 最佳实践建议

我来为你介绍几种实现动态分组聚合的方法,包含 Python、SQL、JavaScript 等常用技术方案。

Python 动态分组聚合

基础实现（使用 pandas）

import pandas as pd
import numpy as np
from typing import List, Dict, Any, Callable
class DynamicGroupAgg:
    """
    动态分组聚合类
    """
    @staticmethod
    def aggregate(
        data: pd.DataFrame,
        group_cols: List[str],
        agg_config: Dict[str, List[str or Callable]]
    ) -> pd.DataFrame:
        """
        动态分组聚合
        Args:
            data: 输入数据
            group_cols: 分组列名列表
            agg_config: 聚合配置，格式：{列名: [聚合函数]}
        Returns:
            聚合结果
        """
        return data.groupby(group_cols).agg(agg_config).reset_index()
    @staticmethod
    def aggregate_with_conditions(
        data: pd.DataFrame,
        group_cols: List[str],
        agg_config: Dict[str, List[str or Callable]],
        conditions: Dict[str, Any] = None
    ) -> pd.DataFrame:
        """
        带条件的动态分组聚合
        """
        if conditions:
            for col, val in conditions.items():
                if isinstance(val, tuple):
                    data = data[data[col].between(*val)]
                else:
                    data = data[data[col] == val]
        return data.groupby(group_cols).agg(agg_config).reset_index()
# 使用示例
if __name__ == "__main__":
    # 示例数据
    df = pd.DataFrame({
        'date': ['2024-01-01', '2024-01-01', '2024-01-02', '2024-01-02'],
        'category': ['A', 'B', 'A', 'B'],
        'sales': [100, 200, 150, 250],
        'quantity': [10, 20, 15, 25]
    })
    # 动态分组配置
    group_cols = ['date']  # 分组列
    agg_config = {
        'sales': ['sum', 'mean', 'max'],
        'quantity': ['sum', 'mean']
    }
    # 执行聚合
    aggregator = DynamicGroupAgg()
    result = aggregator.aggregate(df, group_cols, agg_config)
    print(result)

高级实现（支持自定义函数）

import pandas as pd
from typing import Union, Callable, List, Dict
import json
class AdvancedDynamicAgg:
    """高级动态分组聚合"""
    def __init__(self):
        self.agg_functions = {
            'sum': pd.Series.sum,
            'mean': pd.Series.mean,
            'max': pd.Series.max,
            'min': pd.Series.min,
            'count': pd.Series.count,
            'std': pd.Series.std,
            'var': pd.Series.var,
            'first': lambda x: x.iloc[0],
            'last': lambda x: x.iloc[-1],
            'nunique': pd.Series.nunique,
            'custom': None  # 自定义函数
        }
    def add_custom_function(self, name: str, func: Callable):
        """添加自定义聚合函数"""
        self.agg_functions[name] = func
    def dynamic_aggregate(
        self,
        data: pd.DataFrame,
        group_config: Dict[str, Any],
        agg_config: Dict[str, Dict[str, Any]]
    ) -> pd.DataFrame:
        """
        高级动态分组聚合
        Args:
            data: 输入数据
            group_config: 分组配置 {列名: [级别]}
            agg_config: 聚合配置 {列名: {函数: 参数}}
        """
        # 构建分组
        group_cols = list(group_config.keys())
        groups = data.groupby(group_cols)
        # 构建聚合结果
        result_data = []
        for name, group in groups:
            row = {}
            # 添加分组信息
            if len(group_cols) == 1:
                row[group_cols[0]] = name
            else:
                for i, col in enumerate(group_cols):
                    row[col] = name[i]
            # 执行聚合
            for col, func_config in agg_config.items():
                for func_name, params in func_config.items():
                    if func_name in self.agg_functions:
                        func = self.agg_functions[func_name]
                        if func:
                            row[f"{col}_{func_name}"] = func(group[col])
            result_data.append(row)
        return pd.DataFrame(result_data)
# 使用示例
agg = AdvancedDynamicAgg()
# 添加自定义函数
def custom_sum_with_threshold(series, threshold=100):
    return series[series > threshold].sum()
agg.add_custom_function('sum_threshold', custom_sum_with_threshold)
# 执行聚合
result = agg.dynamic_aggregate(
    df,
    group_config={'category': 'top', 'date': 'all'},
    agg_config={
        'sales': {
            'sum': {},
            'custom': {'threshold': 120}
        }
    }
)

SQL 动态分组聚合

使用动态 SQL

-- 创建存储过程实现动态分组聚合
CREATE OR REPLACE PROCEDURE dynamic_group_aggregate(
    IN table_name VARCHAR(100),
    IN group_cols TEXT,
    IN agg_columns JSON
)
LANGUAGE plpgsql
AS $$
DECLARE
    sql_query TEXT;
    agg_part TEXT;
    col_name TEXT;
    func_name TEXT;
    agg_config JSON;
BEGIN
    -- 构建聚合部分
    agg_part := '';
    FOR agg_config IN SELECT * FROM json_array_elements(agg_columns)
    LOOP
        col_name := agg_config->>'column';
        func_name := agg_config->>'function';
        IF agg_part != '' THEN
            agg_part := agg_part || ', ';
        END IF;
        agg_part := agg_part || func_name || '(' || col_name || ') as ' || 
                    col_name || '_' || func_name;
    END LOOP;
    -- 构建完整SQL
    sql_query := 'SELECT ' || group_cols || ', ' || agg_part || 
                 ' FROM ' || table_name || 
                 ' GROUP BY ' || group_cols;
    -- 执行查询
    EXECUTE sql_query;
END;
$$;
-- 使用示例
CALL dynamic_group_aggregate(
    'sales_data',
    'category, date',
    '[{"column": "amount", "function": "SUM"}, 
      {"column": "amount", "function": "AVG"},
      {"column": "quantity", "function": "COUNT"}]'
);

JavaScript 实现

前端动态聚合

class DynamicGroupAggregator {
  constructor(data) {
    this.data = data;
  }
  /**
   * 动态分组聚合
   * @param {string[]} groupKeys - 分组键
   * @param {Object} aggConfig - 聚合配置
   * @param {Object} filters - 过滤条件
   */
  aggregate(groupKeys, aggConfig, filters = {}) {
    let filteredData = this.applyFilters(filters);
    return this.groupAndAggregate(filteredData, groupKeys, aggConfig);
  }
  applyFilters(filters) {
    let filteredData = [...this.data];
    for (const [key, value] of Object.entries(filters)) {
      if (typeof value === 'function') {
        filteredData = filteredData.filter(value);
      } else if (Array.isArray(value)) {
        filteredData = filteredData.filter(item => 
          value.includes(item[key])
        );
      } else {
        filteredData = filteredData.filter(item => 
          item[key] === value
        );
      }
    }
    return filteredData;
  }
  groupAndAggregate(data, groupKeys, aggConfig) {
    // 创建分组
    const groups = {};
    data.forEach(item => {
      const groupKey = groupKeys.map(key => item[key]).join('|');
      if (!groups[groupKey]) {
        groups[groupKey] = {
          keys: groupKeys.map(key => item[key]),
          items: []
        };
      }
      groups[groupKey].items.push(item);
    });
    // 执行聚合
    const results = [];
    for (const [key, group] of Object.entries(groups)) {
      const result = {};
      // 添加分组信息
      groupKeys.forEach((k, index) => {
        result[k] = group.keys[index];
      });
      // 计算聚合值
      for (const [col, funcs] of Object.entries(aggConfig)) {
        for (const func of funcs) {
          const values = group.items.map(item => item[col]);
          result[`${col}_${func}`] = this.calculateAggregation(values, func);
        }
      }
      results.push(result);
    }
    return results;
  }
  calculateAggregation(values, func) {
    switch (func) {
      case 'sum':
        return values.reduce((a, b) => a + b, 0);
      case 'avg':
        return values.reduce((a, b) => a + b, 0) / values.length;
      case 'max':
        return Math.max(...values);
      case 'min':
        return Math.min(...values);
      case 'count':
        return values.length;
      case 'nunique':
        return new Set(values).size;
      default:
        throw new Error(`Unknown function: ${func}`);
    }
  }
}
// 使用示例
const data = [
  { date: '2024-01-01', category: 'A', sales: 100 },
  { date: '2024-01-01', category: 'B', sales: 200 },
  { date: '2024-01-02', category: 'A', sales: 150 },
  { date: '2024-01-02', category: 'B', sales: 250 }
];
const aggregator = new DynamicGroupAggregator(data);
const result = aggregator.aggregate(
  ['date'],
  {
    'sales': ['sum', 'avg', 'max'],
    'category': ['nunique']
  },
  { sales: val => val > 100 } // 过滤条件
);
console.log(result);

配置驱动的通用实现

import json
import pandas as pd
from typing import Dict, Any
class ConfigDrivenAggregation:
    """配置驱动的聚合系统"""
    def __init__(self, config_file: str = None):
        self.config = {}
        if config_file:
            self.load_config(config_file)
    def load_config(self, config_file: str):
        """加载配置文件"""
        with open(config_file, 'r') as f:
            self.config = json.load(f)
    def set_config(self, config: Dict[str, Any]):
        """设置配置"""
        self.config = config
    def execute(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        执行配置中的聚合操作
        """
        result = data.copy()
        for step in self.config.get('steps', []):
            step_type = step.get('type')
            if step_type == 'filter':
                result = self._apply_filter(result, step)
            elif step_type == 'group_by':
                result = self._apply_group_by(result, step)
            elif step_type == 'aggregate':
                result = self._apply_aggregate(result, step)
            elif step_type == 'merge':
                result = self._apply_merge(result, step)
        return result
    def _apply_filter(self, data: pd.DataFrame, config: Dict) -> pd.DataFrame:
        """应用过滤条件"""
        conditions = config.get('conditions', [])
        for condition in conditions:
            col = condition['column']
            op = condition['operator']
            value = condition['value']
            if op == 'eq':
                data = data[data[col] == value]
            elif op == 'gt':
                data = data[data[col] > value]
            elif op == 'lt':
                data = data[data[col] < value]
            elif op == 'in':
                data = data[data[col].isin(value)]
        return data
    def _apply_group_by(self, data: pd.DataFrame, config: Dict) -> pd.DataFrame:
        """应用分组"""
        columns = config.get('columns', [])
        return data.groupby(columns)
    def _apply_aggregate(self, data: pd.DataFrame, config: Dict) -> pd.DataFrame:
        """应用聚合"""
        # 检查是否已经分组
        if isinstance(data.index, pd.MultiIndex):
            groups = data
        else:
            group_cols = config.get('group_by', [])
            groups = data.groupby(group_cols) if group_cols else data.groupby(lambda x: True)
        # 构建聚合配置
        agg_config = config.get('aggregations', {})
        return groups.agg(agg_config).reset_index()
# 配置文件示例 (config.json)
config_example = {
    "steps": [
        {
            "type": "filter",
            "conditions": [
                {"column": "sales", "operator": "gt", "value": 100}
            ]
        },
        {
            "type": "group_by",
            "columns": ["date", "category"]
        },
        {
            "type": "aggregate",
            "aggregations": {
                "sales": ["sum", "mean"],
                "quantity": ["sum", "count"]
            }
        }
    ]
}
# 使用示例
aggregator = ConfigDrivenAggregation()
aggregator.set_config(config_example)
result = aggregator.execute(df)

最佳实践建议

性能优化

# 使用向量化操作
def optimized_aggregate(data: pd.DataFrame, config: Dict) -> pd.DataFrame:
    """优化的聚合实现"""
    # 使用categorical类型提升性能
    for col in config.get('group_by', []):
        if col in data.columns:
            data[col] = data[col].astype('category')
    # 使用numba加速（需要安装numba）
    from numba import jit
    @jit(nopython=True)
    def fast_sum(values):
        return values.sum()
    # 使用groupby的transform方法
    return data.groupby(config['group_by']).transform('sum')

错误处理

def safe_aggregate(data: pd.DataFrame, config: Dict) -> pd.DataFrame:
    """安全的聚合实现"""
    try:
        # 验证配置
        required_keys = ['group_by', 'aggregations']
        for key in required_keys:
            if key not in config:
                raise ValueError(f"Missing required config key: {key}")
        # 验证列是否存在
        for col in config['group_by']:
            if col not in data.columns:
                raise ValueError(f"Column '{col}' not found in data")
        # 执行聚合
        return data.groupby(config['group_by']).agg(config['aggregations'])
    except Exception as e:
        print(f"Aggregation failed: {e}")
        return pd.DataFrame()

这些实现提供了灵活的动态分组聚合方案,可以根据具体需求选择适合的方式，核心思路是通过配置驱动，支持运行时动态指定分组列和聚合函数。