Akka.NET Adventskalender – Tür 19

Datenmengen schrittweise in Griff bekommen

Gestern haben wir uns mit den Grundlagen von Map/Reduce befasst und die Vorteile dieses Algorithmus bei großen Datenmengen kennengelernt. Heute werden wir uns an unserem konstruierten Beispiel ansehen, welche Schritte notwendig sind.

Als erstes benötigen wir eine zentrale Instanz. Nennen wir ihn Master. Er koordiniert alle anderen Beteiligten, verteilt und sammelt wieder ein.

using System;
using Akka.Actor;
using MapReduce.Actors;
using MapReduce.Messages;

namespace MapReduce.Actors
{
    public class Master : ReceiveActor
    {
        IActorRef mapper;
        IActorRef reducer;
        IActorRef aggregator;

        public Master()
        {
            mapper = Context.ActorOf(Props.Create<Mapper>());
            reducer = Context.ActorOf(Props.Create<Reducer>());
            aggregator = Context.ActorOf(Props.Create<Aggregator>());

            // 1. forward a string to mapper
            Receive<string>(mapper.Tell);

            // 2. forward map result to reducer
            Receive<MapResult>(reducer.Tell);

            // 3. forward reduce result to aggregator
            Receive<ReduceResult>(aggregator.Tell);

            // allow asking for aggregated result at any time
            Receive<GetResult>(aggregator.Forward);
        }
    }
}

Als nächstes definieren wir die diversen Nachrichten, die von den diversen Beteiligten gesendet und empfangen werden.

In Objekten der Klasse LanguageCount wollen wir Tupel der Form (Sprache, Anzahl) festhalten. Wir werden das öfter benötigen.

namespace MapReduce.Messages
{
    public class LanguageCount
    {
        public string Language { get; set; }
        public int Count { get; set; }

        public LanguageCount(string language, int count = 1)
        {
            Language = language;
            Count = count;
        }
    }
}

Der Map Schritt erzeugt Listen solcher Tupel:

using System;
using System.Collections.Generic;
using System.Linq;

namespace MapReduce.Messages
{
    public class MapResult
    {
        public List<LanguageCount> Counts { get; set; }

        public MapResult()
        {
            Counts = new List<LanguageCount>();
        }

        public override string ToString()
        {
            return string.Format("[MapResult: {0}]", 
                String.Join(", ", 
                    Counts.Select(c => String.Format("{0}:{1}", 
                        c.Language, c.Count))
                )
            );
        }
    }
}

Wie schon erwähnt wird diese Nachricht dann an den Reduce Schritt übergeben, der dann eine nach Sprache gruppierte Liste erzeugt. Das ist mit einer Dictionary in .NET wunderbar abbildbar.

using System;
using System.Collections.Generic;
using System.Linq;

namespace MapReduce.Messages
{
    public class ReduceResult
    {
        public Dictionary<string,int> Result { get; set; }

        public ReduceResult()
        {
            Result = new Dictionary<string,int>();
        }

        public override string ToString()
        {
            return string.Format("[ReduceResult: {0}]", 
                String.Join(", ", 
                    Result.Keys.Select(l => String.Format("{0}:{1}", l, Result[l]))
                )
            );
        }

    }
}

Mit diesen Hilfsmitteln ausgestattet werden wir morgen die notwendige Logik programmieren.