Sascha Dittmann Über .NET, SQL Server und die Cloud
Im heutigen Teil meiner kleinen Serie der MapReduce Entwurfsmustern für Microsoft HDInsight dreht es sich erneut um die Berechnung der Standardabweichung und des Medians.
Diesmal werde ich den vorhandenen Algorithmus dahingehend erweitern, dass die Nutzung eines Combiners ermöglicht wird…
Im vergangene Teil dieser Serie hatte ich bereits eine vereinfachtes Beispiel zur Berechnung der Standardabweichung und des Medians vorgestellt.
Das Problem mit dem vorgestellten Algorithmus ist allerdings das er keinen Combiner verwenden kann und dadurch – bei größeren Datenmengen – recht resourcenhungrig wird.
Beispiel "Standardabweichung und Median (mit Combiner)"
In der Mapper-Funktion hat sich im Vergleich zum vergangene Teil nichts geändert.
Diese extrahiert weiterhin das CreationData– und Text-Attribut und gibt die Stunde und Zeichenlänge als Key/Value-Pair zurück:
| public class MedianStdDevMapper : MapperBase | |
| { | |
| public override void Map(string inputLine, MapperContext context) | |
| { | |
| var parsed = XmlUtils.ParseXml(inputLine); | |
| if (parsed == null | |
| || !parsed.ContainsKey("CreationDate") | |
| || !parsed.ContainsKey("Text")) | |
| { | |
| context.CoreContext.IncrementCounter( | |
| "Median / Std. Dev. Mapper", "Invalid Rows", 1); | |
| return; | |
| } | |
| DateTime creationDate; | |
| if (!DateTime.TryParse(parsed["CreationDate"], out creationDate)) | |
| { | |
| context.CoreContext.IncrementCounter( | |
| "Median / Std. Dev. Mapper", "Invalid Creation Dates", 1); | |
| return; | |
| } | |
| var text = parsed["Text"]; | |
| context.EmitKeyValue( | |
| creationDate.Hour.ToString(CultureInfo.InvariantCulture), | |
| text.Length.ToString(CultureInfo.InvariantCulture)); | |
| } | |
| } |
by
Um die Datenmenge zu verkleinern, aggregiert die Combiner-Funktion die erhaltenen Values vor.
Gleiche Textlängen werden hierbei mit ihrer jeweiligen Anzahl als neue Values zurückgegeben.
| public class MedianStdDevCombiner | |
| : JsonOutReducerCombinerBase<MedianStdDevData> | |
| { | |
| public override void Reduce(string key, | |
| IEnumerable<string> values, | |
| JsonReducerCombinerContext<MedianStdDevData> context) | |
| { | |
| var query = values | |
| .Select(int.Parse) | |
| .GroupBy(v => v) | |
| .Select(grp => new MedianStdDevData | |
| { | |
| Value = grp.Key, | |
| Count = grp.Count(), | |
| }); | |
| foreach (var value in query) | |
| context.EmitKeyValue(key, value); | |
| } | |
| } |
Bei der Combiner-Funktion nutze ich erneut eine POCO-Klasse, sowie eine JSON-Basisklasse, um mir die Datenübertragung zu vereinfachen.
| public class MedianStdDevData | |
| { | |
| public int Value { get; set; } | |
| public int Count { get; set; } | |
| } |
Bei der Reducer-Funktion müssen die komprimierten Values wieder "ausgepackt" werden, um anschließend den Median, die Varianz, sowie die Standardabweichung berechnen zu können.
| public class MedianStdDevReducer | |
| : JsonInReducerCombinerBase<MedianStdDevData> | |
| { | |
| public override void Reduce(string key, | |
| IEnumerable<MedianStdDevData> values, | |
| ReducerCombinerContext context) | |
| { | |
| float sum = 0; | |
| long totalComments = 0; | |
| var commentLengthCounts = new Dictionary<int, long>(); | |
| foreach (var data in values) | |
| { | |
| totalComments += data.Count; | |
| sum += data.Value * data.Count; | |
| if (!commentLengthCounts.ContainsKey(data.Value)) | |
| commentLengthCounts.Add(data.Value, data.Count); | |
| else | |
| commentLengthCounts[data.Value] += data.Count; | |
| } | |
| // calculate median | |
| double median = 0; | |
| var medianIndex = totalComments / 2; | |
| long previousComments = 0; | |
| var prevKey = 0; | |
| foreach (var entry in commentLengthCounts.OrderBy(e => e.Key)) | |
| { | |
| if (previousComments <= medianIndex | |
| && medianIndex < previousComments + entry.Value) | |
| { | |
| if (totalComments % 2 == 0 && previousComments == medianIndex) | |
| median = (entry.Key + prevKey) / 2.0f; | |
| else | |
| median = entry.Key; | |
| break; | |
| } | |
| previousComments += entry.Value; | |
| prevKey = entry.Key; | |
| } | |
| // calculate standard deviation | |
| var avg = sum / totalComments; | |
| var sumOfSquares = commentLengthCounts | |
| .Sum(entry => Math.Pow(entry.Key - avg, 2) * entry.Value); | |
| var stdDev = Math.Sqrt(sumOfSquares / (totalComments - 1)); | |
| context.EmitKeyValue( | |
| key, | |
| String.Format("{0}t{1}", median, stdDev)); | |
| } | |
| } |
