Выполнение задачи 4 "MapReduce"

README.md

@@ -0,0 +1,88 @@
+
+## Ключевые классы
+
+| Класс              | Описание |
+|--------------------|----------|
+| **Coordinator**   | Центральный менеджер всего процесса. <br/>— Хранит очередь Map/Reduce задач <br/>— Выдаёт Worker'ам задачи через `synchronized + wait()` <br/>— Отслеживает прогресс через счётчики <br/>— После завершения Map-фазы создаёт Reduce-задачи <br/>— Уведомляет Worker'ов с помощью `notifyAll()` |
+| **Worker**        | Исполнитель задачи (Thread). <br/>— Запрашивает задачу у Coordinator <br/>— Выполняет `task.execute()` <br/>— Завершает работу при получении `null` |
+| **MapTask**       | Задача первой фазы. <br/>— Читает входной файл <br/>— Применяет Mapper <br/>— Разбивает слова по reduce-бакетам: `hash(word) % numReduce` <br/>— Создаёт промежуточные файлы `mr-X-Y.txt` <br/>— Отчитывается координатору о завершении |
+| **ReduceTask**    | Задача второй фазы. <br/>— Читает все промежуточные файлы для своего reduceId <br/>— Группирует данные по ключам <br/>— Сортирует ключи <br/>— Применяет Reducer (подсчёт значений) <br/>— Пишет вывод в `output/mr-out-Y.txt` |
+| **Mapper / Reducer** | Интерфейсы пользовательской логики. <br/>— **WordCountMapper**: разбивает текст на слова, нормализует, выдаёт пары (слово, "1") <br/>— **WordCountReducer**: суммирует значения списка ("1" + "1" + ...) |
+| **KeyValue**      | Простая структура данных (ключ + значение) |
+
+---
+
+## 📂 Работа программы
+
+При запуске `MyWorksApplication` происходит следующее:
+
+1. Создаются 7 входных текстовых файлов (`file1.txt` … `file7.txt`).
+2. Создаются директории:
+    - `intermediate/` — промежуточные файлы map-фазы
+    - `output/` — финальные файлы reduce-фазы
+3. Запускаются worker-потоки.
+4. Map-фаза создаёт файлы вида `mr-X-Y.txt`.
+5. Reduce-фаза собирает и агрегирует данные в `mr-out-Y.txt`.
+6. Программа завершает работу.
+
+---
+
+
+
+
+
+<p>
+Для удобства предоставлю итоговые аутпуты здесь (текст файлов-источников в main методе программы).
+</p>
+
+<h3>📄 mr-out-0.txt</h3>
+<pre><code>are 2
+baby 1
+can 2
+care 1
+i 7
+know 3
+love 2
+me 4
+more 1
+say 1</code></pre>
+
+<h3>📄 mr-out-1.txt</h3>
+<pre><code>a 1
+and 2
+but 1
+do 1
+give 2
+go 1
+hurt 2
+is 4
+just 1
+not 1
+ohh 2
+one 1
+sign 1
+so 1
+up 1
+we 1
+what 5
+you 5</code></pre>
+
+<h3>📄 mr-out-2.txt</h3>
+<pre><code>cant 1
+dont 5
+else 1
+it 1
+my 1
+no 2
+oh 3
+on 1
+ooh 2
+right 1
+there 1
+to 1
+whoa 12
+why 1
+wrong 1</code></pre>
+
+
+

file1.txt

@@ -0,0 +1 @@
+What is love? Baby dont hurt me, dont hurt me, no more

file2.txt

@@ -0,0 +1 @@
+No I dont know why you are not there. I give you my love, but you dont care

file3.txt

@@ -0,0 +1 @@
+So what is right and what is wrong? Give me a sign

file4.txt

@@ -0,0 +1 @@
+Whoa, whoa, whoa, whoa, oh-whoa, whoa, ohh, ooh

file5.txt

@@ -0,0 +1 @@
+Whoa, whoa, whoa, whoa, oh-whoa, whoa, ohh, ooh

file6.txt

@@ -0,0 +1 @@
+Oh, I dont know, what can I do? What else can I say, it is up to you

file7.txt

@@ -0,0 +1 @@
+I know we are one. Just me and you. I cant go on

intermediate/mr-0-0.txt

@@ -0,0 +1,5 @@
+love	1
+baby	1
+me	1
+me	1
+more	1

intermediate/mr-0-1.txt

@@ -0,0 +1,4 @@
+what	1
+is	1
+hurt	1
+hurt	1

intermediate/mr-0-2.txt

@@ -0,0 +1,3 @@
+dont	1
+dont	1
+no	1

intermediate/mr-1-0.txt

@@ -0,0 +1,6 @@
+i	1
+know	1
+are	1
+i	1
+love	1
+care	1

intermediate/mr-1-1.txt

@@ -0,0 +1,6 @@
+you	1
+not	1
+give	1
+you	1
+but	1
+you	1

intermediate/mr-1-2.txt

@@ -0,0 +1,6 @@
+no	1
+dont	1
+why	1
+there	1
+my	1
+dont	1

intermediate/mr-2-0.txt

@@ -0,0 +1 @@
+me	1

intermediate/mr-2-1.txt

@@ -0,0 +1,9 @@
+so	1
+what	1
+is	1
+and	1
+what	1
+is	1
+give	1
+a	1
+sign	1

intermediate/mr-2-2.txt

@@ -0,0 +1,2 @@
+right	1
+wrong	1

intermediate/mr-3-1.txt

@@ -0,0 +1 @@
+ohh	1

intermediate/mr-3-2.txt

@@ -0,0 +1,8 @@
+whoa	1
+whoa	1
+whoa	1
+whoa	1
+oh	1
+whoa	1
+whoa	1
+ooh	1

intermediate/mr-4-1.txt

@@ -0,0 +1 @@
+ohh	1

intermediate/mr-4-2.txt

@@ -0,0 +1,8 @@
+whoa	1
+whoa	1
+whoa	1
+whoa	1
+oh	1
+whoa	1
+whoa	1
+ooh	1

intermediate/mr-5-0.txt

@@ -0,0 +1,7 @@
+i	1
+know	1
+can	1
+i	1
+can	1
+i	1
+say	1

intermediate/mr-5-1.txt

@@ -0,0 +1,6 @@
+what	1
+do	1
+what	1
+is	1
+up	1
+you	1

intermediate/mr-5-2.txt

@@ -0,0 +1,5 @@
+oh	1
+dont	1
+else	1
+it	1
+to	1

intermediate/mr-6-0.txt

@@ -0,0 +1,5 @@
+i	1
+know	1
+are	1
+me	1
+i	1

intermediate/mr-6-1.txt

@@ -0,0 +1,6 @@
+we	1
+one	1
+just	1
+and	1
+you	1
+go	1

intermediate/mr-6-2.txt

@@ -0,0 +1,2 @@
+cant	1
+on	1

output/mr-out-0.txt

@@ -0,0 +1,10 @@
+are 2
+baby 1
+can 2
+care 1
+i 7
+know 3
+love 2
+me 4
+more 1
+say 1

output/mr-out-1.txt

@@ -0,0 +1,18 @@
+a 1
+and 2
+but 1
+do 1
+give 2
+go 1
+hurt 2
+is 4
+just 1
+not 1
+ohh 2
+one 1
+sign 1
+so 1
+up 1
+we 1
+what 5
+you 5

output/mr-out-2.txt

@@ -0,0 +1,15 @@
+cant 1
+dont 5
+else 1
+it 1
+my 1
+no 2
+oh 3
+on 1
+ooh 2
+right 1
+there 1
+to 1
+whoa 12
+why 1
+wrong 1

src/main/java/com/daniilkhanukov/spring/myworks/Coordinator.java

@@ -0,0 +1,104 @@
+package com.daniilkhanukov.spring.myworks;
+
+import com.daniilkhanukov.spring.myworks.map.Mapper;
+import com.daniilkhanukov.spring.myworks.reduce.Reducer;
+import com.daniilkhanukov.spring.myworks.task.MapTask;
+import com.daniilkhanukov.spring.myworks.task.ReduceTask;
+import com.daniilkhanukov.spring.myworks.task.Task;
+
+import java.util.*;
+
+/**
+ * Coordinator:
+ * - хранит очередь задач (taskQueue)
+ * - сначала в очереди находятся MAP задачи
+ * - после того, как все map завершены, coordinator создаёт и кладёт в очередь REDUCE задачи
+ * - метод getTask() блокируется (wait) пока задачи не появятся или пока весь процесс не завершится
+ */
+public class Coordinator {
+    private final Queue<Task> taskQueue = new ArrayDeque<>();
+    private final int numMapTasks;
+    private final int numReduceTasks;
+    private final Reducer reducer;
+
+    // для каждого reduce-id список имён промежуточных файлов, которые нужно обработать
+    private final List<List<String>> intermediateFilesPerReduce;
+
+    private int remainingMapTasks;
+    private int remainingReduceTasks;
+    // флаг завершения
+    private boolean done = false;
+
+    public Coordinator(List<String> inputFiles, int numReduceTasks, Mapper mapper, Reducer reducer) {
+        this.numMapTasks = inputFiles.size();
+        this.remainingMapTasks = numMapTasks;
+        this.numReduceTasks = numReduceTasks;
+        this.remainingReduceTasks = numReduceTasks;
+        this.reducer = reducer;
+
+        // создание списка списков
+        this.intermediateFilesPerReduce = new ArrayList<>(numReduceTasks);
+        for (int i = 0; i < numReduceTasks; i++) {
+            intermediateFilesPerReduce.add(new ArrayList<>());
+        }
+
+        // заполнение taskQueue объектами MapTask
+        for (int mapId = 0; mapId < inputFiles.size(); mapId++) {
+            String fileName = inputFiles.get(mapId);
+            taskQueue.add(new MapTask(mapId, fileName, numReduceTasks, mapper));
+        }
+    }
+
+    /**
+     * Воркеры вызывают этот метод, чтобы получить задачу.
+     * Если нет задач и процесс не завершён - поток ждёт (wait).
+     * Если процесс завершён - возвращает null (сигнал на выход)
+     */
+    public synchronized Task getTask() throws InterruptedException {
+        while (taskQueue.isEmpty() && !done) {
+            wait();
+        }
+        if (done) return null;
+        return taskQueue.poll();
+    }
+
+    /**
+     * Воркеры сообщают о завершении map задачи и передают имена созданных файлов (map: reduceId -> filename).
+     * Если для какого-то reduce файл не создан - в мапе этого reduceId не будет
+     */
+    @SuppressWarnings("java:S1172")
+    public synchronized void reportMapDone(int mapId, Map<Integer, String> createdFiles) {
+        // добавление имён файлов к соответствующим reduce спискам
+        for (Map.Entry<Integer, String> e : createdFiles.entrySet()) {
+            int reduceId = e.getKey();
+            String fileName = e.getValue();
+            intermediateFilesPerReduce.get(reduceId).add(fileName);
+        }
+
+        remainingMapTasks--;
+        // если все map завершены - создаются reduce задачи и будятся воркеры
+        if (remainingMapTasks == 0) {
+
+            for (int r = 0; r < numReduceTasks; r++) {
+                List<String> filesForR = new ArrayList<>(intermediateFilesPerReduce.get(r));
+                taskQueue.add(new ReduceTask(r, filesForR, reducer));
+            }
+            notifyAll();
+        }
+    }
+
+    /**
+     * Воркеры сообщают о завершении reduce задачи.
+     * Когда все reduce выполнены, меняется флаг done=true и пробуждаются ожидающие
+     */
+    @SuppressWarnings("java:S1172")
+    public synchronized void reportReduceDone(int reduceId) {
+        remainingReduceTasks--;
+        if (remainingReduceTasks == 0) {
+            done = true;
+            notifyAll();
+        } else {
+            notifyAll();
+        }
+    }
+}

src/main/java/com/daniilkhanukov/spring/myworks/KeyValue.java

@@ -0,0 +1,24 @@
+package com.daniilkhanukov.spring.myworks;
+
+public class KeyValue {
+    private final String key;
+    private final String value;
+
+    public KeyValue(String key, String value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public String toString() {
+        return key + ": " + value;
+    }
+}