MapReduce Entwurfsmuster – Numerische Aggregation (Average)

AverageCombiner.cs

public class AverageCombiner 
  : JsonInOutReducerCombinerBase<CountAverageData, CountAverageData>
{
  public override void Reduce(string key, 
    IEnumerable<CountAverageData> values, 
    JsonReducerCombinerContext<CountAverageData> context)
  {
    float sum = 0;
    float count = 0;

    foreach (var value in values)
    {
      sum += value.Count * value.Average;
      count += value.Count;
    }

    context.EmitKeyValue(key, new CountAverageData
    {
      Average = sum / count,
      Count = count,
    });
  }
}

AverageMapper.cs

public class AverageMapper 
  : JsonOutMapperBase<CountAverageData>
{
  public override void Map(string inputLine, 
    JsonMapperContext<CountAverageData> context)
  {
    var parsed = XmlUtils.ParseXml(inputLine);

    if (parsed == null 
        || !parsed.ContainsKey("CreationDate") 
        || !parsed.ContainsKey("Text"))
    {
      context.CoreContext.IncrementCounter(
        "Average Mapper", "Invalid Rows", 1);
      return;
    }

    DateTime creationDate;
    if (!DateTime.TryParse(parsed["CreationDate"], out creationDate))
    {
      context.CoreContext.IncrementCounter(
        "Average Mapper", "Invalid Creation Dates", 1);
      return;
    }

    var text = parsed["Text"];

    context.EmitKeyValue(
      creationDate.Hour.ToString(CultureInfo.InvariantCulture),
      new CountAverageData
      {
        Average = text.Length,
        Count = 1,
      });
  }
}

AverageReducer.cs

public class AverageReducer
  : JsonInReducerCombinerBase<CountAverageData>
{
  public override void Reduce(string key, 
    IEnumerable<CountAverageData> values, 
    ReducerCombinerContext context)
  {
    float sum = 0;
    float count = 0;

    foreach (var value in values)
    {
      sum += value.Count * value.Average;
      count += value.Count;
    }

    context.EmitKeyValue(key, new CountAverageData
    {
      Average = sum / count,
      Count = count,
    }.ToString());
  }
}

CountAverageData.cs

public class CountAverageData
{
  public float Average { get; set; }
  public float Count { get; set; }

  public override string ToString()
  {
    return String.Format("{0}t{1}", Average, Count);
  }
}