-
Notifications
You must be signed in to change notification settings - Fork 9
/
PhraseMap.java
70 lines (57 loc) · 2.37 KB
/
PhraseMap.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package phrasecount;
import java.util.Iterator;
import java.util.Optional;
import com.google.common.base.Function;
import com.google.common.collect.Iterators;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.observer.Observer.Context;
import org.apache.fluo.recipes.export.Export;
import org.apache.fluo.recipes.export.ExportQueue;
import org.apache.fluo.recipes.map.CollisionFreeMap;
import org.apache.fluo.recipes.map.Combiner;
import org.apache.fluo.recipes.map.Update;
import org.apache.fluo.recipes.map.UpdateObserver;
import phrasecount.pojos.Counts;
import static phrasecount.Constants.EXPORT_QUEUE_ID;
/**
* This class contains all of the code related to the {@link CollisionFreeMap} that keeps track of phrase counts.
*/
public class PhraseMap {
/**
* A combiner for the {@link CollisionFreeMap} that stores phrase counts. The
* {@link CollisionFreeMap} calls this combiner when it lazily updates the counts for a phrase.
*/
public static class PcmCombiner implements Combiner<String, Counts> {
@Override
public Optional<Counts> combine(String key, Iterator<Counts> updates) {
Counts sum = new Counts(0, 0);
while (updates.hasNext()) {
sum = sum.add(updates.next());
}
return Optional.of(sum);
}
}
/**
* This class is notified when the {@link CollisionFreeMap} used to store phrase counts updates a
* phrase count. Updates are placed an Accumulo export queue to be exported to the table storing
* phrase counts for query.
*/
public static class PcmUpdateObserver extends UpdateObserver<String, Counts> {
private ExportQueue<String, PhraseExport> pcEq;
@Override
public void init(String mapId, Context observerContext) throws Exception {
pcEq = ExportQueue.getInstance(EXPORT_QUEUE_ID, observerContext.getAppConfiguration());
}
@Override
public void updatingValues(TransactionBase tx, Iterator<Update<String, Counts>> updates) {
Iterator<Export<String, PhraseExport>> exports = Iterators.transform(updates,
new Function<Update<String, Counts>, Export<String, PhraseExport>>() {
@Override
public Export<String, PhraseExport> apply(Update<String, Counts> update) {
return new Export<>(update.getKey(), new PhraseExport(update.getNewValue().get()));
}
});
pcEq.addAll(tx, exports);
}
}
}