BAEL-1164 hzaelcast Jet.

2018-08-07 13:14:21 +05:30
parent 2f0b1b46c9
commit 0be91964df
3 changed files with 80 additions and 12 deletions
@@ -0,0 +1,51 @@
+package com.baeldung.hazelcast.jet;
+
+import java.util.List;
+import java.util.Map;
+
+import static com.hazelcast.jet.Traversers.traverseArray;
+import static com.hazelcast.jet.aggregate.AggregateOperations.counting;
+import static com.hazelcast.jet.function.DistributedFunctions.wholeItem;
+
+import com.hazelcast.jet.Jet;
+import com.hazelcast.jet.JetInstance;
+import com.hazelcast.jet.pipeline.Pipeline;
+import com.hazelcast.jet.pipeline.Sinks;
+import com.hazelcast.jet.pipeline.Sources;
+
+public class WordCounter {
+
+    private static final String LIST_NAME = "textList";
+
+    private static final String MAP_NAME = "countMap";
+
+    private Pipeline createPipeLine() {
+        Pipeline p = Pipeline.create();
+        p.drawFrom(Sources.<String> list(LIST_NAME))
+            .flatMap(word -> traverseArray(word.toLowerCase()
+                .split("\\W+")))
+            .filter(word -> !word.isEmpty())
+            .groupingKey(wholeItem())
+            .aggregate(counting())
+            .drainTo(Sinks.map(MAP_NAME));
+        return p;
+    }
+
+    public Long countWord(List<String> sentences, String word) {
+        long count = 0;
+        JetInstance jet = Jet.newJetInstance();
+        try {
+            List<String> textList = jet.getList(LIST_NAME);
+            textList.addAll(sentences);
+            Pipeline p = createPipeLine();
+            jet.newJob(p)
+                .join();
+            Map<String, Long> counts = jet.getMap(MAP_NAME);
+            count = counts.get(word);
+        } finally {
+            Jet.shutdownAll();
+        }
+        return count;
+    }
+
+}