如何基于NCDC气象数据编写reducer.py计算平均温度

需求背景 #

需将原本用于计算温度最大值的reducer脚本，调整为按分组key计算对应温度平均值的逻辑。

NCDC气象数据样本 #

0057011060999991928010112004+67500+012067FM-12+001199999V0202001N012319999999N0500001N9+00281+99999102171ADDAY181999GF108991999999999999001001MD1710261+9999MW1801
0062011060999991928010206004+67500+012067FM-12+001199999V0201801N00931220001CN0200001N9+00281+99999100901ADDAA199002091AY121999GF101991999999017501999999MD1810461+9999
0108011060999991928010212004+67500+012067FM-12+001199999V0201601N009319999999N0100001N9+00111+99999100062ADDAY171999GF108991999011012501001001MD1810542+9999MW1681EQDQ01+000042SCOTLCQ02+100063APOSLPQ03+000542APC3  
0087011060999991928010306004+67500+012067FM-12+001199999V0202001N022619999999N0100001N9+00501+99999098781ADDAA199001091AY161999GF108991999011004501001001MD1310061+9999MW1601EQDQ01+000042SCOTLC
0057011060999991928010312004+67500+012067FM-12+001199999V0202301N01541004501CN0040001N9+00001+99999098951ADDAY161999GF108991081061004501999999MD1210201+9999MW1601

现有计算最大值的Reducer代码 #

#!/usr/bin/env python

import sys

(last_key, max_val) = (None, -sys.maxint)
for line in sys.stdin:
  (key, val) = line.strip().split("\t")
  if last_key and last_key != key:
    print "%s\t%s" % (last_key, max_val)
    (last_key, max_val) = (key, int(val))
  else:
    (last_key, max_val) = (key, max(max_val, int(val)))

if last_key:
  print "%s\t%s" % (last_key, max_val)

修改后的计算平均温度的Reducer代码 #

#!/usr/bin/env python

import sys

# 初始化变量：记录当前分组key、该组温度总和、数据条数
(last_key, sum_temp, count) = (None, 0, 0)

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    key, val = line.split("\t")
    try:
        temp = int(val)
    except ValueError:
        # 跳过无效的温度值
        continue
    
    if last_key and last_key != key:
        # 切换分组时，计算并输出上一组的平均值
        avg_temp = sum_temp / float(count) if count != 0 else 0
        print "%s\t%.2f" % (last_key, avg_temp)
        # 重置当前分组的统计变量
        last_key = key
        sum_temp = temp
        count = 1
    else:
        # 同一分组，累加温度和计数
        last_key = key
        sum_temp += temp
        count += 1

# 处理最后一个分组的数据
if last_key and count != 0:
    avg_temp = sum_temp / float(count)
    print "%s\t%.2f" % (last_key, avg_temp)

修改说明 #

• 替换原本跟踪最大值的变量，改为记录温度总和和数据条数两个核心统计值
• 新增异常处理，跳过无法转换为整数的无效温度数据
• 分组切换时，通过总和/条数计算平均值，保留两位小数输出
• 单独处理最后一组未输出的数据，避免遗漏
• 使用float(count)确保除法运算得到浮点型结果，避免整数除法丢失精度

内容的提问来源于stack exchange，提问作者dStudent