Simple Java Map / Reduce framework [zamknięty]

Sprawdziłeś Akka ? Podczas gdy akka jest tak naprawdę rozproszonym modelem aktora opartym na współbieżności, możesz zaimplementować wiele rzeczy po prostu za pomocą małego kodu. Jest to po prostu tak łatwe do podziału pracy na kawałki z nim, a to automatycznie w pełni wykorzystuje maszynę wielordzeniową, a także jest w stanie korzystać z wielu maszyn do przetwarzania pracy. W przeciwieństwie do używania nici, wydaje mi się to bardziej naturalne.

Mam Java map reduce example używając akka. To nie jest najprostsza Mapa redukuje przykład, ponieważ korzysta z futures; ale powinna dać przybliżony obraz tego, co jest związane. Jest kilka ważnych rzeczy, które pokazuje mój przykład Map reduce:

• Jak podzielić pracę.
• Jak przypisać pracę: akka ma bardzo prosty system wiadomości był również parter pracy, którego harmonogram można skonfigurować. Kiedy nauczyłem się go używać, nie mogłem przestać. To takie proste i elastyczne. Używałem wszystkich czterech rdzeni procesora w mgnieniu oka. Jest to naprawdę świetne do wdrażania usług.
• Jak wiedzieć, kiedy praca jest wykonana, a wynik jest gotowy do przetworzenia: jest to właściwie część, która może być najtrudniejsza i najbardziej myląca do zrozumienia, chyba że jesteś już zaznajomiony z Futures. Nie musisz używać kontraktów Futures, ponieważ istnieją inne opcje. Użyłem ich, bo chciałem czegoś krótszego dla ludzi.

Jeśli masz jakieś pytania, StackOverflow rzeczywiście ma awesome Akka QA sekcja.

Używam następującej struktury

int procs = Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(procs);

List<Future<TaskResult>> results = new ArrayList();
for(int i=0;i<tasks;i++)
    results.add(es.submit(new Task(i)));
for(Future<TaskResult> future:results)
    reduce(future);

Zdaję sobie sprawę, że może to być trochę po fakcie, ale możesz chcieć rzucić okiem na Jsr166y forkjoin klasy z JDK7.

Istnieje wstecznie przeportowana biblioteka, która działa pod JDK6 bez żadnych problemów, więc nie musisz czekać do następnego tysiąclecia, aby ją uruchomić. Znajduje się gdzieś pomiędzy surowym executorem a hadoopem, dając framework do pracy nad zadaniem Map reduce w bieżącym JVM.

Kilka lat temu stworzyłem dla siebie jednorazówkę, kiedy dostałem 8-rdzeniową maszynę, ale nie byłem z niej zbyt zadowolony. Nigdy nie było tak proste w użyciu, jak miałem nadzieję, a zadania wymagające dużej pamięci nie były dobrze skalowane.

Jeśli nie otrzymasz żadnych prawdziwych odpowiedzi mogę podzielić się więcej, ale sednem tego jest:

public class LocalMapReduce<TMapInput, TMapOutput, TOutput> {
    private int m_threads;
    private Mapper<TMapInput, TMapOutput> m_mapper;
    private Reducer<TMapOutput, TOutput> m_reducer;
    ...
    public TOutput mapReduce(Iterator<TMapInput> inputIterator) {
        ExecutorService pool = Executors.newFixedThreadPool(m_threads);
        Set<Future<TMapOutput>> futureSet = new HashSet<Future<TMapOutput>>();
        while (inputIterator.hasNext()) {
            TMapInput m = inputIterator.next();
            Future<TMapOutput> f = pool.submit(m_mapper.makeWorker(m));
            futureSet.add(f);
            Thread.sleep(10);
        }
        while (!futureSet.isEmpty()) {
            Thread.sleep(5);
            for (Iterator<Future<TMapOutput>> fit = futureSet.iterator(); fit.hasNext();) {
                Future<TMapOutput> f = fit.next();
                if (f.isDone()) {
                    fit.remove();
                    TMapOutput x = f.get();
                    m_reducer.reduce(x);
                }
            }
        }
        return m_reducer.getResult();
    }
}

EDIT: na podstawie komentarza, poniżej JEST WERSJA BEZ sleep. Sztuczka polega na użyciu CompletionService, która zasadniczo zapewnia blokowanie kolejki zakończonych Futures.

 public class LocalMapReduce<TMapInput, TMapOutput, TOutput> {
    private int m_threads;
    private Mapper<TMapInput, TMapOutput> m_mapper;
    private Reducer<TMapOutput, TOutput> m_reducer;
    ...
    public TOutput mapReduce(Collection<TMapInput> input) {
        ExecutorService pool = Executors.newFixedThreadPool(m_threads);
        CompletionService<TMapOutput> futurePool = 
                  new ExecutorCompletionService<TMapOutput>(pool);
        Set<Future<TMapOutput>> futureSet = new HashSet<Future<TMapOutput>>();
        for (TMapInput m : input) {
            futureSet.add(futurePool.submit(m_mapper.makeWorker(m)));
        }
        pool.shutdown();
        int n = futureSet.size();
        for (int i = 0; i < n; i++) {
            m_reducer.reduce(futurePool.take().get());
        }
        return m_reducer.getResult();
    }

Zwrócę również uwagę, że jest to bardzo wydestylowany algorytm map-reduce, w tym pojedynczy pracownik reduce, który wykonuje zarówno operację reduce, jak i merge.

Lubię używać Skandium do równoległości w Javie. Framework implementuje pewne wzorce równoległości (mianowicie Master-Slave, Map/Reduce, Pipe, Fork i Divide & Conquer) dla maszyn wielordzeniowych z pamięcią współdzieloną. Technika ta nazywana jest "szkieletami algorytmicznymi". Wzory można zagnieżdżać.

W szczegółach są szkielety i mięśnie. Mięśnie wykonują rzeczywistą pracę(split, merge, execute i condition). Szkielety reprezentują wzory równoległości, z wyjątkiem "While", " For "I " If", co może być przydatne podczas zagnieżdżania wzorców.

Przykłady można znaleźć wewnątrz frameworka. Potrzebowałem trochę, aby zrozumieć, jak korzystać z mięśni i szkieletów, ale po pokonaniu tej przeszkody naprawdę lubię ten framework. :)

Spojrzałeś na GridGain ?

Możesz zajrzeć na stronę projektu Functionals 4 Java: http://f4j.rethab.ch / wprowadza filtr, mapowanie i redukcję do wersji java przed 8.

API MapReduce zostało wprowadzone do wersji 3.2 Hazelcast (zobacz sekcję API MapReduce w dokumentach ). Chociaż Hazelcast jest przeznaczony do stosowania w systemie rozproszonym, działa doskonale w konfiguracji pojedynczego węzła i jest dość lekki.

Możesz wypróbować LeoTask: równoległe działanie zadania i Framework agregacji wyników

Jest darmowy i open-source: https://github.com/mleoking/leotask

Oto krótkie wprowadzenie pokazujące jego API: https://github.com/mleoking/leotask/blob/master/leotask/introduction.pdf?raw=true

Jest to lekki framework pracujący na jednym komputerze z wykorzystaniem wszystkich dostępnych rdzeni procesora.

Posiada następujące cechy:

• Automatyczny & parallel parameter space exploration
• elastyczna i oparta na konfiguracji agregacja wyników
• model programowania skupiający się tylko na kluczowej logice
• niezawodne i automatyczne odzyskiwanie przerw

I Narzędzia:

• dynamiczne i klonowalne struktury sieciowe.
• integracja z Gnuplot
• generowanie sieci według wspólnych modeli sieci
• DelimitedReader: zaawansowany czytnik, który bada CSV (rozdzielany przecinkami wartości) pliki takie jak baza danych
• szybki generator liczb losowych oparty na algorytmie Mersenne Twister
• zintegrowany krzywizna z projektu ImageJ