GCS存储桶JSON数据聚合函数异常，寻求修正方案

需求说明 #

GCS存储桶中有若干JSON文件，数据结构示例如下：

[
  {
    "Id": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
    "Name": "alibaba",
    "storeid": "Y1",
    "storeName": "alibaba1",
    "a": "1/2/3",
    "b": "1.0/1.0/3",
    "c": "0/0/0",
    "d": "0/0/0",
    "e": "1.8/3.4",
    "f": "1/2/3",
    "g": "1/2/3"
  },
  {
    "Id": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
    "Name": "alibaba",
    "storeUuid": "Y2",
    "storeName": "alibaba2",
    "a": "1/2/3",
    "b": "1.0/1.0/3",
    "c": "0/0/0",
    "d": "0/0/0",
    "e": "1.7/2.4",
    "f": "1/2/3",
    "g": "1/2/3"
  },
  {
    "Id": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
    "Name": "alibaba",
    "storeUuid": "Y3",
    "storeName": "alibaba3",
    "a": "1/2/3",
    "b": "1.0/1.0/3",
    "c": "0/0/0",
    "d": "0/0/0",
    "e": "2.7/4.4",
    "f": "1/2/3",
    "g": "1/2/3"
  },
  {
    "Id": "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
    "Name": "alibaba",
    "storeUuid": "Y4",
    "storeName": "alibaba4",
    "a": "1/2/3",
    "b": "1.0/1.0/3",
    "c": "0/0/0",
    "d": "0/0/0",
    "e": "3.7/5.4",
    "f": "1/2/3",
    "g": "1/2/3"
  }
]

需要完成的数据聚合规则：

• 字段a、b、c、d、f、g：按斜杠分割的每个位置分别求和
• 字段e：按斜杠分割的每个位置分别求平均值
• 异常处理：分割值为NaN或e字段值为data unavailable的情况

原代码问题排查 #

原代码存在以下核心问题：

1. 未区分聚合规则：所有字段统一使用求和逻辑，未对e字段单独处理平均值
2. 异常处理不完整：仅处理了data unavailable字符串，未处理分割值为NaN的情况；且e字段遇到data unavailable时直接设为0，会影响平均值计算精度（应排除无效值）
3. 字段长度兼容问题：若不同JSON对象的同一字段分割后长度不一致，会导致numpy数组转为object类型，无法正确执行聚合计算

修正后的代码 #

import numpy as np

def format_large_numbers_optimized(value):
    abs_values = np.abs(value)
    mask = abs_values >= 1e6
    # 处理NaN值，避免格式化出错
    value = np.where(np.isnan(value), 0.0, value)
    formatted_values = np.where(
        mask,
        np.char.add(np.round(value / 1e6, 2).astype(str), "M"),
        np.round(value, 2).astype(str)
    )
    return formatted_values

def process_json_data_optimized(json_list):
    result = {}
    # 明确指定需要处理的目标字段，避免因JSON结构差异漏处理
    target_keys = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
    for key in target_keys:
        result[key] = {'values': []}
    
    # 收集所有目标字段的值
    for json_data in json_list:
        for key in target_keys:
            value = json_data.get(key, '0')
            result[key]['values'].append(value)
    
    for key in target_keys:
        all_values_processed = []
        max_length = 0
        
        # 先处理所有值，确定最大分割长度，统一补0对齐
        for value in result[key]['values']:
            processed = []
            if isinstance(value, str):
                if value == 'data unavailable':
                    # e字段遇到该值时标记为NaN，后续计算平均时排除
                    processed = [np.nan] if key == 'e' else [0.0]
                elif '/' in value:
                    parts = value.split('/')
                    for part in parts:
                        try:
                            num = float(part)
                            # 处理NaN字符串
                            processed.append(num if not np.isnan(num) else 0.0 if key != 'e' else np.nan)
                        except ValueError:
                            processed.append(0.0 if key != 'e' else np.nan)
                    max_length = max(max_length, len(processed))
                else:
                    # 单值字符串转换
                    try:
                        num = float(value)
                        processed.append(num if not np.isnan(num) else 0.0 if key != 'e' else np.nan)
                    except ValueError:
                        processed.append(0.0 if key != 'e' else np.nan)
            elif isinstance(value, (float, int)):
                processed.append(value if not np.isnan(value) else 0.0 if key != 'e' else np.nan)
            else:
                processed.append(0.0 if key != 'e' else np.nan)
            
            all_values_processed.append(processed)
        
        # 统一补0对齐长度
        aligned_values = []
        for vals in all_values_processed:
            aligned = vals + [0.0 if key != 'e' else np.nan] * (max_length - len(vals))
            aligned_values.append(aligned)
        
        numeric_values = np.array(aligned_values, dtype=np.float64)
        
        # 根据字段类型执行聚合
        if key == 'e':
            # 计算平均值：排除NaN值，若所有值都是NaN则返回0
            aggregated = np.nanmean(numeric_values, axis=0)
            # 处理全NaN的情况
            aggregated = np.where(np.isnan(aggregated), 0.0, aggregated)
        else:
            # 求和：自动忽略NaN（已转为0）
            aggregated = np.sum(numeric_values, axis=0)
        
        # 格式化结果并存储
        formatted = '/'.join(format_large_numbers_optimized(aggregated))
        result[key]['summed'] = formatted
    
    # 组装最终结果
    processed_result = {key: data['summed'] for key, data in result.items()}
    processed_result['Id'] = json_list[0]['Id']
    processed_result['Name'] = json_list[0]['Name']
    return processed_result

修正说明 #

1. 明确目标字段：直接指定target_keys，避免因JSON结构差异漏处理字段
2. 区分聚合逻辑：对e字段使用np.nanmean计算平均值，自动排除NaN值；其他字段使用求和
3. 完善异常处理：
   • 处理NaN字符串，转换为numpy的NaN（e字段）或0（其他字段）
   • e字段遇到data unavailable时标记为NaN，不参与平均值计算
4. 长度对齐：统一所有字段分割后的长度，补0（或NaN）避免numpy数组类型错误
5. NaN值兜底：聚合后若出现全NaN的情况，自动转为0，避免格式化出错

内容的提问来源于stack exchange，提问作者Serge de Gosson de Varennes