You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluo.apache.org by kt...@apache.org on 2017/05/03 21:47:27 UTC
[1/3] incubator-fluo-recipes git commit: Updated ExportQ and CFM to
use new ObserverProvider API
Repository: incubator-fluo-recipes
Updated Branches:
refs/heads/master 0cda5ae9c -> 70a4239c8
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
index 0a05919..4c7f86b 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
@@ -16,23 +16,14 @@
package org.apache.fluo.recipes.core.map;
import java.io.Serializable;
-import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.Set;
-import java.util.regex.Pattern;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterators;
-import com.google.common.collect.Sets;
import com.google.common.hash.Hashing;
import org.apache.fluo.api.client.SnapshotBase;
import org.apache.fluo.api.client.TransactionBase;
@@ -42,52 +33,63 @@ import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.api.data.Bytes.BytesBuilder;
import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumn;
import org.apache.fluo.api.data.RowColumnValue;
import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.observer.Observer;
+import org.apache.fluo.api.observer.Observer.NotificationType;
+import org.apache.fluo.api.observer.ObserverProvider;
+import org.apache.fluo.api.observer.StringObserver;
+import org.apache.fluo.recipes.core.combine.CombineQueue;
import org.apache.fluo.recipes.core.common.TableOptimizations;
import org.apache.fluo.recipes.core.common.TableOptimizations.TableOptimizationsFactory;
-import org.apache.fluo.recipes.core.common.RowRange;
-import org.apache.fluo.recipes.core.common.TransientRegistry;
import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
/**
* See the project level documentation for information about this recipe.
*
* @since 1.0.0
+ * @deprecated since 1.1.0 use {@link CombineQueue}
*/
public class CollisionFreeMap<K, V> {
- private static final String UPDATE_RANGE_END = ":u:~";
-
- private static final String DATA_RANGE_END = ":d:~";
-
private Bytes updatePrefix;
private Bytes dataPrefix;
- private Column notifyColumn;
- private Class<K> keyType;
private Class<V> valType;
private SimpleSerializer serializer;
private Combiner<K, V> combiner;
UpdateObserver<K, V> updateObserver;
- private long bufferSize;
static final Column UPDATE_COL = new Column("u", "v");
static final Column NEXT_COL = new Column("u", "next");
private int numBuckets = -1;
+ private CombineQueue<K, V> combineQ;
+ private Observer combineQueueObserver;
+
+ private static class CfmRegistry implements ObserverProvider.Registry {
+
+ Observer observer;
+
+ @Override
+ public void register(Column observedColumn, NotificationType ntfyType, Observer observer) {
+ this.observer = observer;
+ }
+
+ @Override
+ public void registers(Column observedColumn, NotificationType ntfyType, StringObserver observer) {
+ this.observer = observer;
+ }
+ }
+
@SuppressWarnings("unchecked")
- CollisionFreeMap(Options opts, SimpleSerializer serializer) throws Exception {
+ CollisionFreeMap(SimpleConfiguration appConfig, Options opts, SimpleSerializer serializer)
+ throws Exception {
this.updatePrefix = Bytes.of(opts.mapId + ":u:");
this.dataPrefix = Bytes.of(opts.mapId + ":d:");
- this.notifyColumn = new Column("fluoRecipes", "cfm:" + opts.mapId);
- // TODO defer loading classes
- // TODO centralize class loading
- // TODO try to check type params
+
this.numBuckets = opts.numBuckets;
- this.keyType = (Class<K>) getClass().getClassLoader().loadClass(opts.keyType);
this.valType = (Class<V>) getClass().getClassLoader().loadClass(opts.valueType);
this.combiner =
(Combiner<K, V>) getClass().getClassLoader().loadClass(opts.combinerType).newInstance();
@@ -99,184 +101,28 @@ public class CollisionFreeMap<K, V> {
} else {
this.updateObserver = new NullUpdateObserver<>();
}
- this.bufferSize = opts.getBufferSize();
+
+ combineQ = CombineQueue.getInstance(opts.mapId, appConfig);
+
+ // When this class was deprecated, most of its code was copied to CombineQueue. The following
+ // code is a round about way of using that copied code, with having to make anything in
+ // CombineQueue public.
+ CfmRegistry obsRegistry = new CfmRegistry();
+ combineQ.registerObserver(obsRegistry, i -> this.combiner.combine(i.getKey(), i.iterator()), (
+ tx, changes) -> this.updateObserver.updatingValues(tx, Update.transform(changes)));
+ combineQueueObserver = obsRegistry.observer;
}
private V deserVal(Bytes val) {
return serializer.deserialize(val.toArray(), valType);
}
- private Bytes getKeyFromUpdateRow(Bytes prefix, Bytes row) {
- return row.subSequence(prefix.length(), row.length() - 8);
- }
-
void process(TransactionBase tx, Bytes ntfyRow, Column col) throws Exception {
-
- Preconditions.checkState(ntfyRow.startsWith(updatePrefix));
-
- Bytes nextKey = tx.get(ntfyRow, NEXT_COL);
-
- Span span;
-
- if (nextKey != null) {
- Bytes startRow =
- Bytes.builder(ntfyRow.length() + nextKey.length()).append(ntfyRow).append(nextKey)
- .toBytes();
- Span tmpSpan = Span.prefix(ntfyRow);
- Span nextSpan =
- new Span(new RowColumn(startRow, UPDATE_COL), false, tmpSpan.getEnd(),
- tmpSpan.isEndInclusive());
- span = nextSpan;
- } else {
- span = Span.prefix(ntfyRow);
- }
-
- Iterator<RowColumnValue> iter = tx.scanner().over(span).fetch(UPDATE_COL).build().iterator();
-
- Map<Bytes, List<Bytes>> updates = new HashMap<>();
-
- long approxMemUsed = 0;
-
- Bytes partiallyReadKey = null;
- boolean setNextKey = false;
-
- if (iter.hasNext()) {
- Bytes lastKey = null;
- while (iter.hasNext() && approxMemUsed < bufferSize) {
- RowColumnValue rcv = iter.next();
- Bytes curRow = rcv.getRow();
-
- tx.delete(curRow, UPDATE_COL);
-
- Bytes serializedKey = getKeyFromUpdateRow(ntfyRow, curRow);
- lastKey = serializedKey;
-
- List<Bytes> updateList = updates.get(serializedKey);
- if (updateList == null) {
- updateList = new ArrayList<>();
- updates.put(serializedKey, updateList);
- }
-
- Bytes val = rcv.getValue();
- updateList.add(val);
-
- approxMemUsed += curRow.length();
- approxMemUsed += val.length();
- }
-
- if (iter.hasNext()) {
- RowColumnValue rcv = iter.next();
- Bytes curRow = rcv.getRow();
-
- // check if more updates for last key
- if (getKeyFromUpdateRow(ntfyRow, curRow).equals(lastKey)) {
- // there are still more updates for this key
- partiallyReadKey = lastKey;
-
- // start next time at the current key
- tx.set(ntfyRow, NEXT_COL, partiallyReadKey);
- } else {
- // start next time at the next possible key
- Bytes nextPossible =
- Bytes.builder(lastKey.length() + 1).append(lastKey).append(0).toBytes();
- tx.set(ntfyRow, NEXT_COL, nextPossible);
- }
-
- setNextKey = true;
- } else if (nextKey != null) {
- // clear nextKey
- tx.delete(ntfyRow, NEXT_COL);
- }
- } else if (nextKey != null) {
- tx.delete(ntfyRow, NEXT_COL);
- }
-
- if (nextKey != null || setNextKey) {
- // If not all data was read need to run again in the future. If scanning was started in the
- // middle of the bucket, its possible there is new data before nextKey that still needs to be
- // processed. If scanning stopped before reading the entire bucket there may be data after the
- // stop point.
- tx.setWeakNotification(ntfyRow, col);
- }
-
-
-
- BytesBuilder rowBuilder = Bytes.builder();
- rowBuilder.append(dataPrefix);
- rowBuilder.append(ntfyRow.subSequence(updatePrefix.length(), ntfyRow.length()));
- int rowPrefixLen = rowBuilder.getLength();
-
- Set<Bytes> keysToFetch = updates.keySet();
- if (partiallyReadKey != null) {
- final Bytes prk = partiallyReadKey;
- keysToFetch = Sets.filter(keysToFetch, b -> !b.equals(prk));
- }
- Map<Bytes, Map<Column, Bytes>> currentVals = getCurrentValues(tx, rowBuilder, keysToFetch);
-
- ArrayList<Update<K, V>> updatesToReport = new ArrayList<>(updates.size());
-
- for (Entry<Bytes, List<Bytes>> entry : updates.entrySet()) {
- rowBuilder.setLength(rowPrefixLen);
- Bytes currentValueRow = rowBuilder.append(entry.getKey()).toBytes();
- Bytes currVal =
- currentVals.getOrDefault(currentValueRow, Collections.emptyMap()).get(DATA_COLUMN);
-
- Iterator<V> ui = Iterators.transform(entry.getValue().iterator(), this::deserVal);
-
- K kd = serializer.deserialize(entry.getKey().toArray(), keyType);
-
- if (partiallyReadKey != null && partiallyReadKey.equals(entry.getKey())) {
- // not all updates were read for this key, so requeue the combined updates as an update
- Optional<V> nv = combiner.combine(kd, ui);
- if (nv.isPresent()) {
- update(tx, Collections.singletonMap(kd, nv.get()));
- }
- } else {
- Optional<V> nv = combiner.combine(kd, concat(ui, currVal));
- Bytes newVal = nv.isPresent() ? Bytes.of(serializer.serialize(nv.get())) : null;
- if (newVal != null ^ currVal != null || (currVal != null && !currVal.equals(newVal))) {
- if (newVal == null) {
- tx.delete(currentValueRow, DATA_COLUMN);
- } else {
- tx.set(currentValueRow, DATA_COLUMN, newVal);
- }
-
- Optional<V> cvd = Optional.ofNullable(currVal).map(this::deserVal);
- updatesToReport.add(new Update<>(kd, cvd, nv));
- }
- }
- }
-
- // TODO could clear these as converted to objects to avoid double memory usage
- updates.clear();
- currentVals.clear();
-
- if (updatesToReport.size() > 0) {
- updateObserver.updatingValues(tx, updatesToReport.iterator());
- }
+ combineQueueObserver.process(tx, ntfyRow, col);
}
private static final Column DATA_COLUMN = new Column("data", "current");
- private Map<Bytes, Map<Column, Bytes>> getCurrentValues(TransactionBase tx, BytesBuilder prefix,
- Set<Bytes> keySet) {
-
- Set<Bytes> rows = new HashSet<>();
-
- int prefixLen = prefix.getLength();
- for (Bytes key : keySet) {
- prefix.setLength(prefixLen);
- rows.add(prefix.append(key).toBytes());
- }
-
- try {
- return tx.get(rows, Collections.singleton(DATA_COLUMN));
- } catch (IllegalArgumentException e) {
- System.out.println(rows.size());
- throw e;
- }
- }
-
private Iterator<V> concat(Iterator<V> updates, Bytes currentVal) {
if (currentVal == null) {
return updates;
@@ -296,7 +142,6 @@ public class CollisionFreeMap<K, V> {
int hash = Hashing.murmur3_32().hashBytes(k).asInt();
String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
-
BytesBuilder rowBuilder = Bytes.builder();
rowBuilder.append(updatePrefix).append(bucketId).append(':').append(k);
@@ -339,42 +184,7 @@ public class CollisionFreeMap<K, V> {
* updated. The values in the map will be queued for updating.
*/
public void update(TransactionBase tx, Map<K, V> updates) {
- Preconditions.checkState(numBuckets > 0, "Not initialized");
-
- Set<String> buckets = new HashSet<>();
-
- BytesBuilder rowBuilder = Bytes.builder();
- rowBuilder.append(updatePrefix);
- int prefixLength = rowBuilder.getLength();
-
- byte[] startTs = encSeq(tx.getStartTimestamp());
-
- for (Entry<K, V> entry : updates.entrySet()) {
- byte[] k = serializer.serialize(entry.getKey());
- int hash = Hashing.murmur3_32().hashBytes(k).asInt();
- String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
-
- // reset to the common row prefix
- rowBuilder.setLength(prefixLength);
-
- Bytes row = rowBuilder.append(bucketId).append(':').append(k).append(startTs).toBytes();
- Bytes val = Bytes.of(serializer.serialize(entry.getValue()));
-
- // TODO set if not exists would be comforting here.... but
- // collisions on bucketId+key+uuid should never occur
- tx.set(row, UPDATE_COL, val);
-
- buckets.add(bucketId);
- }
-
- for (String bucketId : buckets) {
- rowBuilder.setLength(prefixLength);
- rowBuilder.append(bucketId).append(':');
-
- Bytes row = rowBuilder.toBytes();
-
- tx.setWeakNotification(row, notifyColumn);
- }
+ combineQ.addAll(tx, updates);
}
static String genBucketId(int bucket, int maxBucket) {
@@ -391,7 +201,7 @@ public class CollisionFreeMap<K, V> {
SimpleConfiguration appConf) {
Options opts = new Options(mapId, appConf);
try {
- return new CollisionFreeMap<>(opts, SimpleSerializer.getInstance(appConf));
+ return new CollisionFreeMap<>(appConf, opts, SimpleSerializer.getInstance(appConf));
} catch (Exception e) {
// TODO
throw new RuntimeException(e);
@@ -413,39 +223,29 @@ public class CollisionFreeMap<K, V> {
* @see CollisionFreeMap#getInitializer(String, int, SimpleSerializer)
*
* @since 1.0.0
+ * @deprecated since 1.1.0
*/
+ @Deprecated
public static class Initializer<K2, V2> implements Serializable {
private static final long serialVersionUID = 1L;
- private Bytes dataPrefix;
-
- private SimpleSerializer serializer;
-
- private int numBuckets = -1;
+ private org.apache.fluo.recipes.core.combine.CombineQueue.Initializer<K2, V2> initializer;
private Initializer(String mapId, int numBuckets, SimpleSerializer serializer) {
- this.dataPrefix = Bytes.of(mapId + ":d:");
- this.numBuckets = numBuckets;
- this.serializer = serializer;
+ this.initializer = CombineQueue.getInitializer(mapId, numBuckets, serializer);
}
public RowColumnValue convert(K2 key, V2 val) {
- byte[] k = serializer.serialize(key);
- int hash = Hashing.murmur3_32().hashBytes(k).asInt();
- String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
-
- BytesBuilder bb = Bytes.builder(dataPrefix.length() + bucketId.length() + 1 + k.length);
- Bytes row = bb.append(dataPrefix).append(bucketId).append(':').append(k).toBytes();
- byte[] v = serializer.serialize(val);
-
- return new RowColumnValue(row, DATA_COLUMN, Bytes.of(v));
+ return initializer.convert(key, val);
}
}
/**
* @since 1.0.0
+ * @deprecated since 1.1.0
*/
+ @Deprecated
public static class Options {
static final long DEFAULT_BUFFER_SIZE = 1 << 22;
@@ -531,7 +331,6 @@ public class CollisionFreeMap<K, V> {
/**
* Sets the number of buckets per tablet to generate. This affects how many split points will be
* generated when optimizing the Accumulo table.
- *
*/
public Options setBucketsPerTablet(int bucketsPerTablet) {
Preconditions.checkArgument(bucketsPerTablet > 0, "bucketsPerTablet is <= 0 : "
@@ -540,14 +339,6 @@ public class CollisionFreeMap<K, V> {
return this;
}
- int getBucketsPerTablet() {
- if (bucketsPerTablet == null) {
- return DEFAULT_BUCKETS_PER_TABLET;
- }
-
- return bucketsPerTablet;
- }
-
public <K, V> Options(String mapId, Class<? extends Combiner<K, V>> combiner, Class<K> keyType,
Class<V> valueType, int buckets) {
this(mapId, combiner.getName(), keyType.getName(), valueType.getName(), buckets);
@@ -561,19 +352,10 @@ public class CollisionFreeMap<K, V> {
}
void save(SimpleConfiguration appConfig) {
- appConfig.setProperty(PREFIX + mapId + ".buckets", numBuckets + "");
appConfig.setProperty(PREFIX + mapId + ".combiner", combinerType + "");
- appConfig.setProperty(PREFIX + mapId + ".key", keyType);
- appConfig.setProperty(PREFIX + mapId + ".val", valueType);
if (updateObserverType != null) {
appConfig.setProperty(PREFIX + mapId + ".updateObserver", updateObserverType + "");
}
- if (bufferSize != null) {
- appConfig.setProperty(PREFIX + mapId + ".bufferSize", bufferSize);
- }
- if (bucketsPerTablet != null) {
- appConfig.setProperty(PREFIX + mapId + ".bucketsPerTablet", bucketsPerTablet);
- }
}
}
@@ -582,20 +364,27 @@ public class CollisionFreeMap<K, V> {
* Fluo.
*/
public static void configure(FluoConfiguration fluoConfig, Options opts) {
+ org.apache.fluo.recipes.core.combine.CombineQueue.FluentOptions cqopts =
+ CombineQueue.configure(opts.mapId).keyType(opts.keyType).valueType(opts.valueType)
+ .buckets(opts.numBuckets);
+ if (opts.bucketsPerTablet != null) {
+ cqopts.bucketsPerTablet(opts.bucketsPerTablet);
+ }
+ if (opts.bufferSize != null) {
+ cqopts.bufferSize(opts.bufferSize);
+ }
+ cqopts.save(fluoConfig);
+
opts.save(fluoConfig.getAppConfiguration());
+
fluoConfig.addObserver(new ObserverSpecification(CollisionFreeMapObserver.class.getName(),
ImmutableMap.of("mapId", opts.mapId)));
-
- Bytes dataRangeEnd = Bytes.of(opts.mapId + DATA_RANGE_END);
- Bytes updateRangeEnd = Bytes.of(opts.mapId + UPDATE_RANGE_END);
-
- new TransientRegistry(fluoConfig.getAppConfiguration()).addTransientRange("cfm." + opts.mapId,
- new RowRange(dataRangeEnd, updateRangeEnd));
-
- TableOptimizations.registerOptimization(fluoConfig.getAppConfiguration(), opts.mapId,
- Optimizer.class);
}
+ /**
+ * @deprecated since 1.1.0 use {@link org.apache.fluo.recipes.core.combine.CombineQueue.Optimizer}
+ */
+ @Deprecated
public static class Optimizer implements TableOptimizationsFactory {
/**
@@ -607,55 +396,8 @@ public class CollisionFreeMap<K, V> {
*/
@Override
public TableOptimizations getTableOptimizations(String mapId, SimpleConfiguration appConfig) {
- Options opts = new Options(mapId, appConfig);
-
- BytesBuilder rowBuilder = Bytes.builder();
- rowBuilder.append(mapId);
-
- List<Bytes> dataSplits = new ArrayList<>();
- for (int i = opts.getBucketsPerTablet(); i < opts.numBuckets; i += opts.getBucketsPerTablet()) {
- String bucketId = genBucketId(i, opts.numBuckets);
- rowBuilder.setLength(mapId.length());
- dataSplits.add(rowBuilder.append(":d:").append(bucketId).toBytes());
- }
- Collections.sort(dataSplits);
-
- List<Bytes> updateSplits = new ArrayList<>();
- for (int i = opts.getBucketsPerTablet(); i < opts.numBuckets; i += opts.getBucketsPerTablet()) {
- String bucketId = genBucketId(i, opts.numBuckets);
- rowBuilder.setLength(mapId.length());
- updateSplits.add(rowBuilder.append(":u:").append(bucketId).toBytes());
- }
- Collections.sort(updateSplits);
-
- Bytes dataRangeEnd = Bytes.of(opts.mapId + DATA_RANGE_END);
- Bytes updateRangeEnd = Bytes.of(opts.mapId + UPDATE_RANGE_END);
-
- List<Bytes> splits = new ArrayList<>();
- splits.add(dataRangeEnd);
- splits.add(updateRangeEnd);
- splits.addAll(dataSplits);
- splits.addAll(updateSplits);
-
- TableOptimizations tableOptim = new TableOptimizations();
- tableOptim.setSplits(splits);
-
- tableOptim.setTabletGroupingRegex(Pattern.quote(mapId + ":") + "[du]:");
-
- return tableOptim;
+ return new org.apache.fluo.recipes.core.combine.CombineQueue.Optimizer()
+ .getTableOptimizations(mapId, appConfig);
}
}
-
- private static byte[] encSeq(long l) {
- byte[] ret = new byte[8];
- ret[0] = (byte) (l >>> 56);
- ret[1] = (byte) (l >>> 48);
- ret[2] = (byte) (l >>> 40);
- ret[3] = (byte) (l >>> 32);
- ret[4] = (byte) (l >>> 24);
- ret[5] = (byte) (l >>> 16);
- ret[6] = (byte) (l >>> 8);
- ret[7] = (byte) (l >>> 0);
- return ret;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMapObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMapObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMapObserver.java
index a8e333d..4bfbe95 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMapObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMapObserver.java
@@ -25,7 +25,9 @@ import org.apache.fluo.api.observer.AbstractObserver;
* CollisionFreeMap.Options) . This class should never have to be used directly.
*
* @since 1.0.0
+ * @deprecated since 1.1.0
*/
+@Deprecated
public class CollisionFreeMapObserver extends AbstractObserver {
@SuppressWarnings("rawtypes")
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Combiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Combiner.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Combiner.java
index 4d3ee77..99f12e0 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Combiner.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Combiner.java
@@ -18,8 +18,12 @@ package org.apache.fluo.recipes.core.map;
import java.util.Iterator;
import java.util.Optional;
+import org.apache.fluo.recipes.core.combine.CombineQueue;
+
/**
* @since 1.0.0
+ * @deprecated since 1.1.0 use {@link org.apache.fluo.recipes.core.combine.Combiner} and
+ * {@link CombineQueue}
*/
public interface Combiner<K, V> {
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/NullUpdateObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/NullUpdateObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/NullUpdateObserver.java
index 7bbfec1..f6abaf6 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/NullUpdateObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/NullUpdateObserver.java
@@ -19,6 +19,7 @@ import java.util.Iterator;
import org.apache.fluo.api.client.TransactionBase;
+@Deprecated
class NullUpdateObserver<K, V> extends UpdateObserver<K, V> {
@Override
public void updatingValues(TransactionBase tx, Iterator<Update<K, V>> updates) {}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Update.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Update.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Update.java
index 1965111..2c0c891 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Update.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/Update.java
@@ -15,11 +15,18 @@
package org.apache.fluo.recipes.core.map;
+import java.util.Iterator;
import java.util.Optional;
+import org.apache.fluo.recipes.core.combine.ChangeObserver.Change;
+
+import com.google.common.collect.Iterators;
+
/**
* @since 1.0.0
+ * @deprecated since 1.1.0
*/
+@Deprecated
public class Update<K, V> {
private final K key;
@@ -43,4 +50,9 @@ public class Update<K, V> {
public Optional<V> getOldValue() {
return oldValue;
}
+
+ static <K2, V2> Iterator<Update<K2, V2>> transform(Iterable<Change<K2, V2>> changes) {
+ return Iterators.transform(changes.iterator(), change -> new Update<K2, V2>(change.getKey(),
+ change.getOldValue(), change.getNewValue()));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/map/UpdateObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/UpdateObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/UpdateObserver.java
index b00e1b3..8046f90 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/UpdateObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/UpdateObserver.java
@@ -19,17 +19,20 @@ import java.util.Iterator;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.observer.Observer.Context;
+import org.apache.fluo.recipes.core.combine.ChangeObserver;
+import org.apache.fluo.recipes.core.combine.CombineQueue;
/**
* A {@link CollisionFreeMap} calls this to allow additional processing to be done when key values
* are updated. See the project level documentation for more information.
*
* @since 1.0.0
+ * @deprecated since 1.1.0 use {@link ChangeObserver} and {@link CombineQueue}
*/
+@Deprecated
public abstract class UpdateObserver<K, V> {
+
public void init(String mapId, Context observerContext) throws Exception {}
public abstract void updatingValues(TransactionBase tx, Iterator<Update<K, V>> updates);
-
- // TODO add close
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/types/TypedObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/types/TypedObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/types/TypedObserver.java
index ca68285..6dc0814 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/types/TypedObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/types/TypedObserver.java
@@ -18,14 +18,14 @@ package org.apache.fluo.recipes.core.types;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.observer.AbstractObserver;
+import org.apache.fluo.api.observer.Observer;
/**
- * An {@link AbstractObserver} that uses a {@link TypeLayer}
+ * An {@link Observer} that uses a {@link TypeLayer}
*
* @since 1.0.0
*/
-public abstract class TypedObserver extends AbstractObserver {
+public abstract class TypedObserver implements Observer {
private final TypeLayer tl;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/OptionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/OptionsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/OptionsTest.java
new file mode 100644
index 0000000..3818d64
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/OptionsTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OptionsTest {
+ @Test
+ public void testExportQueueOptions() {
+ FluoConfiguration conf = new FluoConfiguration();
+
+ CombineQueue.configure("Q1").keyType("KT").valueType("VT").buckets(100).save(conf);
+ CombineQueue.configure("Q2").keyType("KT2").valueType("VT2").buckets(200).bucketsPerTablet(20)
+ .bufferSize(1000000).save(conf);
+
+ SimpleConfiguration appConfig = conf.getAppConfiguration();
+
+ Assert.assertEquals(CqConfigurator.getKeyType("Q1", appConfig), "KT");
+ Assert.assertEquals(CqConfigurator.getValueType("Q1", appConfig), "VT");
+ Assert.assertEquals(CqConfigurator.getNumBucket("Q1", appConfig), 100);
+ Assert.assertEquals(CqConfigurator.getBucketsPerTablet("Q1", appConfig),
+ CqConfigurator.DEFAULT_BUCKETS_PER_TABLET);
+ Assert.assertEquals(CqConfigurator.getBufferSize("Q1", appConfig),
+ CqConfigurator.DEFAULT_BUFFER_SIZE);
+
+ Assert.assertEquals(CqConfigurator.getKeyType("Q2", appConfig), "KT2");
+ Assert.assertEquals(CqConfigurator.getValueType("Q2", appConfig), "VT2");
+ Assert.assertEquals(CqConfigurator.getNumBucket("Q2", appConfig), 200);
+ Assert.assertEquals(CqConfigurator.getBucketsPerTablet("Q2", appConfig), 20);
+ Assert.assertEquals(CqConfigurator.getBufferSize("Q2", appConfig), 1000000);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/SplitsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/SplitsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/SplitsTest.java
new file mode 100644
index 0000000..efb0ef7
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/SplitsTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.recipes.core.combine.CombineQueue.Optimizer;
+import org.apache.fluo.recipes.core.common.TableOptimizations;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SplitsTest {
+ private static List<Bytes> sort(List<Bytes> in) {
+ ArrayList<Bytes> out = new ArrayList<>(in);
+ Collections.sort(out);
+ return out;
+ }
+
+ @Test
+ public void testSplits() {
+
+ FluoConfiguration fluoConfig = new FluoConfiguration();
+
+ CombineQueue.configure("foo").keyType(String.class).valueType(Long.class).buckets(3)
+ .bucketsPerTablet(1).save(fluoConfig);
+
+ TableOptimizations tableOptim1 =
+ new Optimizer().getTableOptimizations("foo", fluoConfig.getAppConfiguration());
+ List<Bytes> expected1 =
+ Lists.transform(
+ Arrays.asList("foo:d:1", "foo:d:2", "foo:d:~", "foo:u:1", "foo:u:2", "foo:u:~"),
+ Bytes::of);
+
+ Assert.assertEquals(expected1, sort(tableOptim1.getSplits()));
+
+ CombineQueue.configure("bar").keyType(String.class).valueType(Long.class).buckets(6)
+ .bucketsPerTablet(2).save(fluoConfig);
+
+ TableOptimizations tableOptim2 =
+ new Optimizer().getTableOptimizations("bar", fluoConfig.getAppConfiguration());
+ List<Bytes> expected2 =
+ Lists.transform(
+ Arrays.asList("bar:d:2", "bar:d:4", "bar:d:~", "bar:u:2", "bar:u:4", "bar:u:~"),
+ Bytes::of);
+ Assert.assertEquals(expected2, sort(tableOptim2.getSplits()));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/it/CombineQueueTreeIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/it/CombineQueueTreeIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/it/CombineQueueTreeIT.java
new file mode 100644
index 0000000..1379948
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/combine/it/CombineQueueTreeIT.java
@@ -0,0 +1,281 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine.it;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.api.observer.ObserverProvider;
+import org.apache.fluo.recipes.core.combine.ChangeObserver.Change;
+import org.apache.fluo.recipes.core.combine.CombineQueue;
+import org.apache.fluo.recipes.core.combine.SummingCombiner;
+import org.apache.fluo.recipes.core.map.it.TestSerializer;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CombineQueueTreeIT {
+
+ private static final String CQ_XYT_ID = "xyt";
+ private static final String CQ_XY_ID = "xy";
+ private static final String CQ_XT_ID = "xt";
+ private static final String CQ_YT_ID = "yt";
+ private static final String CQ_X_ID = "x";
+ private static final String CQ_T_ID = "t";
+ private static final String CQ_Y_ID = "y";
+
+ public static class CqitObserverProvider implements ObserverProvider {
+
+ @Override
+ public void provide(Registry or, Context ctx) {
+ // Link combine queues into a tree of combine queues.
+
+ SimpleConfiguration appCfg = ctx.getAppConfiguration();
+ CombineQueue<String, Long> xytCq = CombineQueue.getInstance(CQ_XYT_ID, appCfg);
+ CombineQueue<String, Long> xyCq = CombineQueue.getInstance(CQ_XY_ID, appCfg);
+ CombineQueue<String, Long> ytCq = CombineQueue.getInstance(CQ_YT_ID, appCfg);
+ CombineQueue<String, Long> xtCq = CombineQueue.getInstance(CQ_XT_ID, appCfg);
+ CombineQueue<String, Long> xCq = CombineQueue.getInstance(CQ_X_ID, appCfg);
+ CombineQueue<String, Long> tCq = CombineQueue.getInstance(CQ_T_ID, appCfg);
+ CombineQueue<String, Long> yCq = CombineQueue.getInstance(CQ_Y_ID, appCfg);
+
+ xytCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ Map<String, Long> xtUpdates = new HashMap<>();
+ Map<String, Long> ytUpdates = new HashMap<>();
+ Map<String, Long> xyUpdates = new HashMap<>();
+ for (Change<String, Long> change : changes) {
+ String[] fields = change.getKey().split(":");
+ long delta = change.getNewValue().orElse(0L) - change.getOldValue().orElse(0L);
+ xtUpdates.merge(fields[0] + ":" + fields[2], delta, Long::sum);
+ ytUpdates.merge(fields[1] + ":" + fields[2], delta, Long::sum);
+ xyUpdates.merge(fields[0] + ":" + fields[1], delta, Long::sum);
+ }
+
+ xtCq.addAll(tx, xtUpdates);
+ ytCq.addAll(tx, ytUpdates);
+ xyCq.addAll(tx, xyUpdates);
+
+ invert("xyt", tx, changes);
+ });
+
+ xyCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ Map<String, Long> xUpdates = new HashMap<>();
+ Map<String, Long> yUpdates = new HashMap<>();
+ for (Change<String, Long> change : changes) {
+ String[] fields = change.getKey().split(":");
+ long delta = change.getNewValue().orElse(0L) - change.getOldValue().orElse(0L);
+ xUpdates.merge(fields[0], delta, Long::sum);
+ yUpdates.merge(fields[1], delta, Long::sum);
+ }
+
+ xCq.addAll(tx, xUpdates);
+ yCq.addAll(tx, yUpdates);
+
+ invert("xy", tx, changes);
+ });
+
+ xtCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ Map<String, Long> tUpdates = new HashMap<>();
+ for (Change<String, Long> change : changes) {
+ String[] fields = change.getKey().split(":");
+ long delta = change.getNewValue().orElse(0L) - change.getOldValue().orElse(0L);
+ tUpdates.merge(fields[1], delta, Long::sum);
+ }
+
+ tCq.addAll(tx, tUpdates);
+
+ invert("xt", tx, changes);
+ });
+
+ ytCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ invert("yt", tx, changes);
+ });
+
+ xCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ invert("x", tx, changes);
+ });
+
+ tCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ invert("t", tx, changes);
+ });
+
+ yCq.registerObserver(or, new SummingCombiner<>(), (tx, changes) -> {
+ invert("y", tx, changes);
+ });
+ }
+
+ private static void invert(String prefix, TransactionBase tx,
+ Iterable<Change<String, Long>> changes) {
+ for (Change<String, Long> change : changes) {
+ change.getOldValue().ifPresent(
+ val -> tx.delete("inv:" + prefix + ":" + val, new Column("inv", change.getKey())));
+ change.getNewValue().ifPresent(
+ val -> tx.set("inv:" + prefix + ":" + val, new Column("inv", change.getKey()), ""));
+ }
+ }
+
+ }
+
+ private MiniFluo miniFluo;
+
+ @Before
+ public void setUpFluo() throws Exception {
+ FileUtils.deleteQuietly(new File("target/mini"));
+
+ FluoConfiguration props = new FluoConfiguration();
+ props.setApplicationName("eqt");
+ props.setWorkerThreads(20);
+ props.setMiniDataDir("target/mini");
+
+ CombineQueue.configure(CQ_XYT_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_XT_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_XY_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_YT_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_X_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_T_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+ CombineQueue.configure(CQ_Y_ID).keyType(String.class).valueType(Long.class).buckets(7)
+ .save(props);
+
+ props.setObserverProvider(CqitObserverProvider.class);
+
+ SimpleSerializer.setSerializer(props, TestSerializer.class);
+
+ miniFluo = FluoFactory.newMiniFluo(props);
+ }
+
+ private static Map<String, Long> merge(Map<String, Long> m1, Map<String, Long> m2) {
+ Map<String, Long> ret = new HashMap<>(m1);
+ m2.forEach((k, v) -> ret.merge(k, v, Long::sum));
+ return ret;
+ }
+
+ private static Map<String, Long> rollup(Map<String, Long> m, String rollupFields) {
+ boolean useX = rollupFields.contains("x");
+ boolean useY = rollupFields.contains("y");
+ boolean useTime = rollupFields.contains("t");
+
+ Map<String, Long> ret = new HashMap<>();
+ m.forEach((k, v) -> {
+ String[] fields = k.split(":");
+ String nk =
+ (useX ? fields[0] : "") + (useY ? ((useX ? ":" : "") + fields[1]) : "")
+ + (useTime ? ((useX || useY ? ":" : "") + fields[2]) : "");
+
+ ret.merge(nk, v, Long::sum);
+ });
+ return ret;
+ }
+
+ private static Map<String, Long> readRollup(Snapshot snap, String rollupFields) {
+ Map<String, Long> ret = new HashMap<>();
+
+ String prefix = "inv:" + rollupFields + ":";
+ for (RowColumnValue rcv : snap.scanner().over(Span.prefix("inv:" + rollupFields + ":")).build()) {
+ String row = rcv.getsRow();
+ long count = Long.valueOf(row.substring(prefix.length(), row.length()));
+ Assert.assertNull(ret.put(rcv.getColumn().getsQualifier(), count));
+ }
+
+ return ret;
+ }
+
+ @Test
+ public void testCqTree() {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+
+ CombineQueue<String, Long> xytCq =
+ CombineQueue.getInstance(CQ_XYT_ID, fc.getAppConfiguration());
+
+ Map<String, Long> updates1 = new HashMap<>();
+ updates1.put("5:4:23", 1L);
+ updates1.put("5:5:23", 1L);
+ updates1.put("7:5:23", 1L);
+ updates1.put("9:2:23", 1L);
+ updates1.put("5:4:29", 1L);
+ updates1.put("6:6:29", 1L);
+ updates1.put("7:5:29", 1L);
+ updates1.put("9:3:29", 1L);
+ updates1.put("3:3:31", 1L);
+
+ try (Transaction tx = fc.newTransaction()) {
+ xytCq.addAll(tx, updates1);
+ tx.commit();
+ }
+
+ miniFluo.waitForObservers();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+ for (String fieldsNames : new String[] {"x", "y", "t", "xy", "xt", "yt", "xyt"}) {
+ Map<String, Long> expected = rollup(updates1, fieldsNames);
+ Map<String, Long> actual = readRollup(snap, fieldsNames);
+
+ Assert.assertEquals(expected, actual);
+ }
+ }
+
+ Map<String, Long> updates2 = new HashMap<>();
+ updates2.put("7:5:23", 1L);
+ updates2.put("6:6:7", 1L);
+ updates2.put("4:3:31", 1L);
+ updates2.put("9:1:23", 1L);
+ updates2.put("1:2:3", 1L);
+
+ try (Transaction tx = fc.newTransaction()) {
+ xytCq.addAll(tx, updates2);
+ tx.commit();
+ }
+
+ miniFluo.waitForObservers();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+ for (String fieldsNames : new String[] {"x", "y", "t", "xy", "xt", "yt", "xyt"}) {
+ Map<String, Long> expected = rollup(merge(updates1, updates2), fieldsNames);
+ Map<String, Long> actual = readRollup(snap, fieldsNames);
+
+ Assert.assertEquals(expected, actual);
+ }
+ }
+ }
+ }
+
+ @After
+ public void tearDownFluo() throws Exception {
+ if (miniFluo != null) {
+ miniFluo.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/common/TestGrouping.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/common/TestGrouping.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/common/TestGrouping.java
index 2159b61..4c1d7f7 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/common/TestGrouping.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/common/TestGrouping.java
@@ -19,33 +19,32 @@ import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import com.google.common.collect.ImmutableSet;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.recipes.core.combine.CombineQueue;
+import org.apache.fluo.recipes.core.combine.CombineQueue.Optimizer;
import org.apache.fluo.recipes.core.export.ExportQueue;
-import org.apache.fluo.recipes.core.map.CollisionFreeMap;
-import org.apache.fluo.recipes.core.map.CollisionFreeMap.Options;
import org.junit.Assert;
import org.junit.Test;
+import com.google.common.collect.ImmutableSet;
+
public class TestGrouping {
@Test
public void testTabletGrouping() {
FluoConfiguration conf = new FluoConfiguration();
- CollisionFreeMap.configure(conf, new Options("m1", "ct", "kt", "vt", 119));
- CollisionFreeMap.configure(conf, new Options("m2", "ct", "kt", "vt", 3));
-
- ExportQueue.configure(conf, new ExportQueue.Options("eq1", "et", "kt", "vt", 7));
- ExportQueue.configure(conf, new ExportQueue.Options("eq2", "et", "kt", "vt", 3));
+ CombineQueue.configure("m1").keyType("kt").valueType("vt").buckets(119).save(conf);
+ CombineQueue.configure("m2").keyType("kt").valueType("vt").buckets(3).save(conf);
+ ExportQueue.configure("eq1").keyType("kt").valueType("vt").buckets(7).save(conf);
+ ExportQueue.configure("eq2").keyType("kt").valueType("vt").buckets(3).save(conf);
SimpleConfiguration appConfg = conf.getAppConfiguration();
- TableOptimizations tableOptim =
- new CollisionFreeMap.Optimizer().getTableOptimizations("m1", appConfg);
- tableOptim.merge(new CollisionFreeMap.Optimizer().getTableOptimizations("m2", appConfg));
+ TableOptimizations tableOptim = new Optimizer().getTableOptimizations("m1", appConfg);
+ tableOptim.merge(new Optimizer().getTableOptimizations("m2", appConfg));
tableOptim.merge(new ExportQueue.Optimizer().getTableOptimizations("eq1", appConfg));
tableOptim.merge(new ExportQueue.Optimizer().getTableOptimizations("eq2", appConfg));
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
index 376223b..2e5ca8b 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
@@ -15,16 +15,19 @@
package org.apache.fluo.recipes.core.export;
+import java.util.List;
+
import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.config.SimpleConfiguration;
-import org.apache.fluo.recipes.core.export.ExportQueue;
import org.apache.fluo.recipes.core.export.ExportQueue.Options;
import org.junit.Assert;
import org.junit.Test;
public class OptionsTest {
@Test
- public void testExportQueueOptions() {
+ @Deprecated
+ public void testDeprecatedExportQueueOptions() {
FluoConfiguration conf = new FluoConfiguration();
SimpleConfiguration ec1 = new SimpleConfiguration();
@@ -37,26 +40,67 @@ public class OptionsTest {
Options opts1 = new Options("Q1", conf.getAppConfiguration());
- Assert.assertEquals(opts1.exporterType, "ET");
- Assert.assertEquals(opts1.keyType, "KT");
- Assert.assertEquals(opts1.valueType, "VT");
- Assert.assertEquals(opts1.numBuckets, 100);
- Assert.assertEquals(opts1.bucketsPerTablet.intValue(), Options.DEFAULT_BUCKETS_PER_TABLET);
- Assert.assertEquals(opts1.bufferSize.intValue(), Options.DEFAULT_BUFFER_SIZE);
+ Assert.assertEquals(opts1.fluentCfg.exporterType, "ET");
+ Assert.assertEquals(opts1.fluentCfg.keyType, "KT");
+ Assert.assertEquals(opts1.fluentCfg.valueType, "VT");
+ Assert.assertEquals(opts1.fluentCfg.buckets, 100);
+ Assert.assertEquals(opts1.fluentCfg.bucketsPerTablet.intValue(),
+ FluentConfigurator.DEFAULT_BUCKETS_PER_TABLET);
+ Assert.assertEquals(opts1.fluentCfg.bufferSize.intValue(),
+ FluentConfigurator.DEFAULT_BUFFER_SIZE);
Options opts2 = new Options("Q2", conf.getAppConfiguration());
- Assert.assertEquals(opts2.exporterType, "ET2");
- Assert.assertEquals(opts2.keyType, "KT2");
- Assert.assertEquals(opts2.valueType, "VT2");
- Assert.assertEquals(opts2.numBuckets, 200);
- Assert.assertEquals(opts2.bucketsPerTablet.intValue(), 20);
- Assert.assertEquals(opts2.bufferSize.intValue(), 1000000);
+ Assert.assertEquals(opts2.fluentCfg.exporterType, "ET2");
+ Assert.assertEquals(opts2.fluentCfg.keyType, "KT2");
+ Assert.assertEquals(opts2.fluentCfg.valueType, "VT2");
+ Assert.assertEquals(opts2.fluentCfg.buckets, 200);
+ Assert.assertEquals(opts2.fluentCfg.bucketsPerTablet.intValue(), 20);
+ Assert.assertEquals(opts2.fluentCfg.bufferSize.intValue(), 1000000);
SimpleConfiguration ec2 = opts2.getExporterConfiguration();
Assert.assertEquals("ev1", ec2.getString("ep1"));
Assert.assertEquals(3, ec2.getInt("ep2"));
+ List<ObserverSpecification> obsSpecs = conf.getObserverSpecifications();
+ Assert.assertTrue(obsSpecs.size() == 2);
+ for (ObserverSpecification ospec : obsSpecs) {
+ Assert.assertEquals(ExportObserver.class.getName(), ospec.getClassName());
+ String qid = ospec.getConfiguration().getString("queueId");
+ Assert.assertTrue(qid.equals("Q1") || qid.equals("Q2"));
+ }
+ }
+
+ @Test
+ public void testExportQueueOptions() {
+ FluoConfiguration conf = new FluoConfiguration();
+
+ SimpleConfiguration ec1 = new SimpleConfiguration();
+ ec1.setProperty("ep1", "ev1");
+ ec1.setProperty("ep2", 3L);
+
+ ExportQueue.configure("Q1").keyType("KT").valueType("VT").buckets(100).save(conf);
+ ExportQueue.configure("Q2").keyType("KT2").valueType("VT2").buckets(200).bucketsPerTablet(20)
+ .bufferSize(1000000).save(conf);
+
+ FluentConfigurator opts1 = FluentConfigurator.load("Q1", conf.getAppConfiguration());
+
+ Assert.assertNull(opts1.exporterType);
+ Assert.assertEquals(opts1.keyType, "KT");
+ Assert.assertEquals(opts1.valueType, "VT");
+ Assert.assertEquals(opts1.buckets, 100);
+ Assert.assertEquals(opts1.bucketsPerTablet.intValue(),
+ FluentConfigurator.DEFAULT_BUCKETS_PER_TABLET);
+ Assert.assertEquals(opts1.bufferSize.intValue(), FluentConfigurator.DEFAULT_BUFFER_SIZE);
+
+ FluentConfigurator opts2 = FluentConfigurator.load("Q2", conf.getAppConfiguration());
+
+ Assert.assertNull(opts2.exporterType);
+ Assert.assertEquals(opts2.keyType, "KT2");
+ Assert.assertEquals(opts2.valueType, "VT2");
+ Assert.assertEquals(opts2.buckets, 200);
+ Assert.assertEquals(opts2.bucketsPerTablet.intValue(), 20);
+ Assert.assertEquals(opts2.bufferSize.intValue(), 1000000);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
index c9765a7..52f51f1 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
@@ -19,34 +19,27 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
-import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.observer.StringObserver;
import org.apache.fluo.recipes.core.export.ExportQueue;
-import org.apache.fluo.recipes.core.export.it.ExportTestBase.RefExporter;
-import org.apache.fluo.recipes.core.types.TypedObserver;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-public class DocumentObserver extends TypedObserver {
+public class DocumentObserver implements StringObserver {
ExportQueue<String, RefUpdates> refExportQueue;
- @Override
- public void init(Context context) throws Exception {
- refExportQueue = ExportQueue.getInstance(RefExporter.QUEUE_ID, context.getAppConfiguration());
- }
+ private static final Column CURRENT_COL = new Column("content", "current");
- @Override
- public ObservedColumn getObservedColumn() {
- return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
+ DocumentObserver(ExportQueue<String, RefUpdates> refExportQueue) {
+ this.refExportQueue = refExportQueue;
}
@Override
- public void process(TypedTransactionBase tx, Bytes row, Column col) {
- String newContent = tx.get().row(row).col(col).toString();
+ public void process(TransactionBase tx, String row, Column col) {
+ String newContent = tx.gets(row, col).toString();
Set<String> newRefs = new HashSet<>(Arrays.asList(newContent.split(" ")));
Set<String> currentRefs =
- new HashSet<>(Arrays.asList(tx.get().row(row).fam("content").qual("current").toString("")
- .split(" ")));
+ new HashSet<>(Arrays.asList(tx.gets(row, CURRENT_COL, "").split(" ")));
Set<String> addedRefs = new HashSet<>(newRefs);
addedRefs.removeAll(currentRefs);
@@ -59,6 +52,6 @@ public class DocumentObserver extends TypedObserver {
refExportQueue.add(tx, key, val);
- tx.mutate().row(row).fam("content").qual("current").set(newContent);
+ tx.set(row, CURRENT_COL, newContent);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
index 3c210a9..48050ed 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
@@ -36,19 +36,21 @@ import org.apache.fluo.api.client.scanner.CellScanner;
import org.apache.fluo.api.client.scanner.ColumnScanner;
import org.apache.fluo.api.client.scanner.RowScanner;
import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.data.ColumnValue;
import org.apache.fluo.api.data.Span;
import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.api.observer.ObserverProvider;
import org.apache.fluo.recipes.core.export.ExportQueue;
-import org.apache.fluo.recipes.core.export.Exporter;
import org.apache.fluo.recipes.core.export.SequencedExport;
+import org.apache.fluo.recipes.core.export.function.Exporter;
import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
+import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG;
+
public class ExportTestBase {
private static Map<String, Map<String, RefInfo>> globalExports = new HashMap<>();
@@ -95,7 +97,7 @@ public class ExportTestBase {
}
}
- public static class RefExporter extends Exporter<String, RefUpdates> {
+ public static class RefExporter implements Exporter<String, RefUpdates> {
public static final String QUEUE_ID = "req";
@@ -105,7 +107,7 @@ public class ExportTestBase {
}
@Override
- protected void processExports(Iterator<SequencedExport<String, RefUpdates>> exportIterator) {
+ public void export(Iterator<SequencedExport<String, RefUpdates>> exportIterator) {
ArrayList<SequencedExport<String, RefUpdates>> exportList = new ArrayList<>();
Iterators.addAll(exportList, exportIterator);
@@ -135,6 +137,18 @@ public class ExportTestBase {
return null;
}
+ public static class ExportTestObserverProvider implements ObserverProvider {
+
+ @Override
+ public void provide(Registry or, Context ctx) {
+ ExportQueue<String, RefUpdates> refExportQueue =
+ ExportQueue.getInstance(RefExporter.QUEUE_ID, ctx.getAppConfiguration());
+
+ or.register(new Column("content", "new"), STRONG, new DocumentObserver(refExportQueue));
+ refExportQueue.registerObserver(or, new RefExporter());
+ }
+ }
+
@Before
public void setUpFluo() throws Exception {
FileUtils.deleteQuietly(new File("target/mini"));
@@ -143,22 +157,18 @@ public class ExportTestBase {
props.setApplicationName("eqt");
props.setWorkerThreads(20);
props.setMiniDataDir("target/mini");
-
- ObserverSpecification doc = new ObserverSpecification(DocumentObserver.class.getName());
- props.addObserver(doc);
+ props.setObserverProvider(ExportTestObserverProvider.class);
SimpleSerializer.setSerializer(props, GsonSerializer.class);
- ExportQueue.Options exportQueueOpts =
- new ExportQueue.Options(RefExporter.QUEUE_ID, RefExporter.class, String.class,
- RefUpdates.class, getNumBuckets());
-
- if (getBufferSize() != null) {
- exportQueueOpts.setBufferSize(getBufferSize());
+ if (getBufferSize() == null) {
+ ExportQueue.configure(RefExporter.QUEUE_ID).keyType(String.class).valueType(RefUpdates.class)
+ .buckets(getNumBuckets()).save(props);
+ } else {
+ ExportQueue.configure(RefExporter.QUEUE_ID).keyType(String.class).valueType(RefUpdates.class)
+ .buckets(getNumBuckets()).bufferSize(getBufferSize()).save(props);
}
- ExportQueue.configure(props, exportQueueOpts);
-
miniFluo = FluoFactory.newMiniFluo(props);
globalExports.clear();
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/OptionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/OptionsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/OptionsTest.java
index 6587f6e..876aa57 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/OptionsTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/OptionsTest.java
@@ -20,6 +20,7 @@ import org.apache.fluo.recipes.core.map.CollisionFreeMap.Options;
import org.junit.Assert;
import org.junit.Test;
+@Deprecated
public class OptionsTest {
@Test
public void testExportQueueOptions() {
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
index 30f93ff..319d89a 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
@@ -29,6 +29,7 @@ import org.apache.fluo.recipes.core.map.it.WordCountCombiner;
import org.junit.Assert;
import org.junit.Test;
+@Deprecated
public class SplitsTest {
private static List<Bytes> sort(List<Bytes> in) {
ArrayList<Bytes> out = new ArrayList<>(in);
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
index 4213cff..86d62df 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
@@ -55,6 +55,8 @@ import org.junit.Test;
* This test configures a small buffer size and verifies that multiple passes are made to process
* updates.
*/
+@Deprecated
+// TODO migrate to CombineQueue test when removing CFM
public class BigUpdateIT {
private static final TypeLayer tl = new TypeLayer(new StringEncoder());
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
index e22ddea..1c6c0b2 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
@@ -41,6 +41,8 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+@Deprecated
+// TODO migrate to CombineQueue test when removing CFM
public class CollisionFreeMapIT {
private MiniFluo miniFluo;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
index c469151..aa048a8 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
@@ -26,6 +26,8 @@ import org.apache.fluo.recipes.core.map.CollisionFreeMap;
import org.apache.fluo.recipes.core.types.TypedObserver;
import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+@Deprecated
+// TODO move to CombineQueue test when removing CFM
public class DocumentObserver extends TypedObserver {
CollisionFreeMap<String, Long> wcm;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
index df1389f..f819c15 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
@@ -20,6 +20,8 @@ import java.util.Optional;
import org.apache.fluo.recipes.core.map.Combiner;
+@Deprecated
+// TODO move to CombineQueue test when removing CFM
public class WordCountCombiner implements Combiner<String, Long> {
@Override
public Optional<Long> combine(String key, Iterator<Long> updates) {
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
index 1230d99..6b0818e 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
@@ -24,6 +24,8 @@ import org.apache.fluo.api.data.Column;
import org.apache.fluo.recipes.core.map.Update;
import org.apache.fluo.recipes.core.map.UpdateObserver;
+@Deprecated
+// TODO move to CombineQueue test when removing CFM
public class WordCountObserver extends UpdateObserver<String, Long> {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
index d4ad914..09200a1 100644
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
+++ b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
@@ -23,14 +23,17 @@ import java.util.Random;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.fluo.api.client.FluoClient;
import org.apache.fluo.api.client.FluoFactory;
import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
+import org.apache.fluo.api.observer.ObserverProvider;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter;
import org.apache.fluo.recipes.core.export.ExportQueue;
import org.apache.fluo.recipes.test.AccumuloExportITBase;
import org.apache.hadoop.io.Text;
@@ -42,6 +45,24 @@ public class AccumuloExporterIT extends AccumuloExportITBase {
private String exportTable;
public static final String QUEUE_ID = "aeqt";
+ public static class AccumuloExporterObserverProvider implements ObserverProvider {
+
+ @Override
+ public void provide(Registry obsRegistry, Context ctx) {
+ SimpleConfiguration appCfg = ctx.getAppConfiguration();
+
+ ExportQueue<String, String> teq = ExportQueue.getInstance(QUEUE_ID, appCfg);
+
+ teq.registerObserver(obsRegistry, new AccumuloExporter<>(QUEUE_ID, appCfg, (export,
+ mutConsumer) -> {
+ Mutation m = new Mutation(export.getKey());
+ m.put("cf", "cq", export.getSequence(), export.getValue());
+ mutConsumer.accept(m);
+ }));
+ }
+
+ }
+
@Override
public void preFluoInitHook() throws Exception {
@@ -51,12 +72,16 @@ public class AccumuloExporterIT extends AccumuloExportITBase {
MiniAccumuloCluster miniAccumulo = getMiniAccumuloCluster();
- ExportQueue.configure(
- getFluoConfiguration(),
- new ExportQueue.Options(QUEUE_ID, SimpleExporter.class, String.class, String.class, 5)
- .setBucketsPerTablet(1).setExporterConfiguration(
- new AccumuloExporter.Configuration(miniAccumulo.getInstanceName(), miniAccumulo
- .getZooKeepers(), ACCUMULO_USER, ACCUMULO_PASSWORD, exportTable)));
+ getFluoConfiguration().setObserverProvider(AccumuloExporterObserverProvider.class);
+
+ ExportQueue.configure(QUEUE_ID).keyType(String.class).valueType(String.class).buckets(5)
+ .bucketsPerTablet(1).save(getFluoConfiguration());
+
+ AccumuloExporter.configure(QUEUE_ID)
+ .instance(miniAccumulo.getInstanceName(), miniAccumulo.getZooKeepers())
+ .credentials(ACCUMULO_USER, ACCUMULO_PASSWORD).table(exportTable)
+ .save(getFluoConfiguration());
+
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
index 635f6e9..c78262e 100644
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
+++ b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
@@ -28,8 +28,9 @@ import org.apache.fluo.api.client.FluoClient;
import org.apache.fluo.api.client.FluoFactory;
import org.apache.fluo.api.client.Transaction;
import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
+import org.apache.fluo.api.observer.ObserverProvider;
import org.apache.fluo.recipes.accumulo.export.AccumuloReplicator;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter;
import org.apache.fluo.recipes.core.export.ExportQueue;
import org.apache.fluo.recipes.core.transaction.TxLog;
import org.apache.fluo.recipes.test.AccumuloExportITBase;
@@ -46,6 +47,16 @@ public class AccumuloReplicatorIT extends AccumuloExportITBase {
public static final String QUEUE_ID = "repq";
private TypeLayer tl = new TypeLayer(new StringEncoder());
+ public static class AritObserverProvider implements ObserverProvider {
+
+ @Override
+ public void provide(Registry or, Context ctx) {
+ ExportQueue<String, TxLog> eq = ExportQueue.getInstance(QUEUE_ID, ctx.getAppConfiguration());
+ eq.registerObserver(or, new AccumuloExporter<>(QUEUE_ID, ctx.getAppConfiguration(),
+ AccumuloReplicator.getTranslator()));
+ }
+ }
+
@Override
public void preFluoInitHook() throws Exception {
@@ -55,11 +66,15 @@ public class AccumuloReplicatorIT extends AccumuloExportITBase {
MiniAccumuloCluster miniAccumulo = getMiniAccumuloCluster();
- ExportQueue.configure(getFluoConfiguration(), new ExportQueue.Options(QUEUE_ID,
- AccumuloReplicator.class.getName(), String.class.getName(), TxLog.class.getName(), 5)
- .setExporterConfiguration(new AccumuloExporter.Configuration(
- miniAccumulo.getInstanceName(), miniAccumulo.getZooKeepers(), ACCUMULO_USER,
- ACCUMULO_PASSWORD, exportTable)));
+ ExportQueue.configure(QUEUE_ID).keyType(String.class).valueType(TxLog.class).buckets(5)
+ .save(getFluoConfiguration());
+
+ AccumuloExporter.configure(QUEUE_ID)
+ .instance(miniAccumulo.getInstanceName(), miniAccumulo.getZooKeepers())
+ .credentials(ACCUMULO_USER, ACCUMULO_PASSWORD).table(exportTable)
+ .save(getFluoConfiguration());
+
+ getFluoConfiguration().setObserverProvider(AritObserverProvider.class);
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
deleted file mode 100644
index d48857e..0000000
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License. You may obtain a
- * copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package org.apache.fluo.recipes.test.export;
-
-import java.util.function.Consumer;
-
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
-import org.apache.fluo.recipes.core.export.SequencedExport;
-
-public class SimpleExporter extends AccumuloExporter<String, String> {
-
- @Override
- protected void translate(SequencedExport<String, String> export, Consumer<Mutation> consumer) {
- Mutation m = new Mutation(export.getKey());
- m.put("cf", "cq", export.getSequence(), export.getValue());
- consumer.accept(m);
- }
-}
[3/3] incubator-fluo-recipes git commit: Updated ExportQ and CFM to
use new ObserverProvider API
Posted by kt...@apache.org.
Updated ExportQ and CFM to use new ObserverProvider API
Created a new CombineQueue API that replaces the CollisionFreeMap. The new
combineQ uses the new Fluo Observer APIs exclusively. The CollisionFreeMap
only uses the old Fluo Observer APIs and was deprecated.
The ExportQueue was modified to support a Fluent configuration mechanism. This
can only be used when using the new Observer API.
Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/commit/70a4239c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/tree/70a4239c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/diff/70a4239c
Branch: refs/heads/master
Commit: 70a4239c837223a287c5eb1a396326ab7cc5be67
Parents: 0cda5ae
Author: Keith Turner <kt...@apache.org>
Authored: Wed Mar 8 18:46:28 2017 -0500
Committer: Keith Turner <kt...@apache.org>
Committed: Wed May 3 17:45:45 2017 -0400
----------------------------------------------------------------------
README.md | 5 +-
docs/accumulo-export-queue.md | 84 ++--
docs/cfm.md | 257 ------------
docs/combine-queue.md | 224 +++++++++++
docs/export-queue.md | 251 ++++++------
.../accumulo/export/AccumuloExporter.java | 69 ++--
.../accumulo/export/AccumuloReplicator.java | 26 +-
.../recipes/accumulo/export/AccumuloWriter.java | 169 --------
.../export/function/AccumuloExporter.java | 104 +++++
.../export/function/AccumuloTranslator.java | 86 ++++
.../export/function/AccumuloWriter.java | 158 ++++++++
.../export/function/AeFluentConfigurator.java | 104 +++++
.../accumulo/export/AccumuloExportTest.java | 111 ------
.../accumulo/export/AccumuloTranslatorTest.java | 112 ++++++
.../fluo/recipes/core/combine/ChangeImpl.java | 48 +++
.../recipes/core/combine/ChangeObserver.java | 52 +++
.../fluo/recipes/core/combine/CombineQueue.java | 183 +++++++++
.../recipes/core/combine/CombineQueueImpl.java | 320 +++++++++++++++
.../fluo/recipes/core/combine/Combiner.java | 51 +++
.../recipes/core/combine/CqConfigurator.java | 148 +++++++
.../fluo/recipes/core/combine/CqOptimizer.java | 70 ++++
.../recipes/core/combine/InitializerImpl.java | 52 +++
.../fluo/recipes/core/combine/InputImpl.java | 68 ++++
.../recipes/core/combine/SummingCombiner.java | 37 ++
.../fluo/recipes/core/export/ExportBucket.java | 1 +
.../fluo/recipes/core/export/ExportEntry.java | 1 +
.../recipes/core/export/ExportObserver.java | 106 +----
.../recipes/core/export/ExportObserverImpl.java | 91 +++++
.../fluo/recipes/core/export/ExportQueue.java | 221 +++++++----
.../fluo/recipes/core/export/Exporter.java | 7 +-
.../recipes/core/export/FluentConfigurator.java | 156 ++++++++
.../recipes/core/export/MemLimitIterator.java | 54 +++
.../recipes/core/export/function/Exporter.java | 87 +++++
.../fluo/recipes/core/map/CollisionFreeMap.java | 388 ++++---------------
.../core/map/CollisionFreeMapObserver.java | 2 +
.../apache/fluo/recipes/core/map/Combiner.java | 4 +
.../recipes/core/map/NullUpdateObserver.java | 1 +
.../apache/fluo/recipes/core/map/Update.java | 12 +
.../fluo/recipes/core/map/UpdateObserver.java | 7 +-
.../fluo/recipes/core/types/TypedObserver.java | 6 +-
.../fluo/recipes/core/combine/OptionsTest.java | 48 +++
.../fluo/recipes/core/combine/SplitsTest.java | 66 ++++
.../core/combine/it/CombineQueueTreeIT.java | 281 ++++++++++++++
.../fluo/recipes/core/common/TestGrouping.java | 21 +-
.../fluo/recipes/core/export/OptionsTest.java | 72 +++-
.../core/export/it/DocumentObserver.java | 27 +-
.../recipes/core/export/it/ExportTestBase.java | 40 +-
.../fluo/recipes/core/map/OptionsTest.java | 1 +
.../fluo/recipes/core/map/SplitsTest.java | 1 +
.../fluo/recipes/core/map/it/BigUpdateIT.java | 2 +
.../recipes/core/map/it/CollisionFreeMapIT.java | 2 +
.../recipes/core/map/it/DocumentObserver.java | 2 +
.../recipes/core/map/it/WordCountCombiner.java | 2 +
.../recipes/core/map/it/WordCountObserver.java | 2 +
.../recipes/test/export/AccumuloExporterIT.java | 39 +-
.../test/export/AccumuloReplicatorIT.java | 27 +-
.../recipes/test/export/SimpleExporter.java | 32 --
57 files changed, 3257 insertions(+), 1341 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index fe511a6..75c105d 100644
--- a/README.md
+++ b/README.md
@@ -31,7 +31,8 @@ like Spark and Kryo.
Recipes are documented below and in the [Recipes API docs][recipes-api].
-* [Collision Free Map][cfm] - A recipe for making many to many updates.
+* [Combine Queue][combine-q] - A recipe for concurrently updating many keys while avoiding
+ collisions.
* [Export Queue][export-q] - A recipe for exporting data from Fluo to external systems.
* [Row Hash Prefix][row-hasher] - A recipe for spreading data evenly in a row prefix.
* [RecordingTransaction][recording-tx] - A wrapper for a Fluo transaction that records all transaction
@@ -102,7 +103,7 @@ Below is a sample Maven POM containing all possible Fluo Recipes dependencies:
[fluo]: https://fluo.apache.org/
[fluo-api]: https://fluo.apache.org/apidocs/fluo/
[recipes-api]: https://fluo.apache.org/apidocs/fluo-recipes/
-[cfm]: docs/cfm.md
+[combine-q]: docs/combine-queue.md
[export-q]: docs/export-queue.md
[recording-tx]: docs/recording-tx.md
[serialization]: docs/serialization.md
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/docs/accumulo-export-queue.md
----------------------------------------------------------------------
diff --git a/docs/accumulo-export-queue.md b/docs/accumulo-export-queue.md
index 218e41e..b037b52 100644
--- a/docs/accumulo-export-queue.md
+++ b/docs/accumulo-export-queue.md
@@ -19,8 +19,8 @@ limitations under the License.
## Background
The [Export Queue Recipe][1] provides a generic foundation for building export mechanism to any
-external data store. The [AccumuloExporter] provides an implementation of this recipe for
-Accumulo. The [AccumuloExporter] is located the `fluo-recipes-accumulo` module and provides the
+external data store. The [AccumuloExporter] provides an [Exporter] for writing to
+Accumulo. [AccumuloExporter] is located in the `fluo-recipes-accumulo` module and provides the
following functionality:
* Safely batches writes to Accumulo made by multiple transactions exporting data.
@@ -32,80 +32,86 @@ following functionality:
Exporting to Accumulo is easy. Follow the steps below:
-1. Implement a class that extends [AccumuloExporter]. This class will process exported objects that
- are placed on your export queue. For example, the `SimpleExporter` class below processes String
- key/value exports and generates mutations for Accumulo.
+1. First, implement [AccumuloTranslator]. Your implementation translates exported
+ objects to Accumulo Mutations. For example, the `SimpleTranslator` class below translates String
+ key/values and into mutations for Accumulo. This step is optional, a lambda could
+ be used in step 3 instead of creating a class.
```java
- public class SimpleExporter extends AccumuloExporter<String, String> {
+ public class SimpleTranslator implements AccumuloTranslator<String,String> {
@Override
- protected void translate(SequencedExport<String, String> export, Consumer<Mutation> consumer) {
+ public void translate(SequencedExport<String, String> export, Consumer<Mutation> consumer) {
Mutation m = new Mutation(export.getKey());
m.put("cf", "cq", export.getSequence(), export.getValue());
consumer.accept(m);
}
}
+
```
-2. With a `SimpleExporter` created, configure an `ExportQueue` to use `SimpleExporter` and
- give it information on how to connect to Accumulo.
+2. Configure an `ExportQueue` and the export table prior to initializing Fluo.
```java
FluoConfiguration fluoConfig = ...;
- // Set accumulo configuration
String instance = // Name of accumulo instance exporting to
String zookeepers = // Zookeepers used by Accumulo instance exporting to
String user = // Accumulo username, user that can write to exportTable
String password = // Accumulo user password
String exportTable = // Name of table to export to
+ // Set properties for table to export to in Fluo app configuration.
+ AccumuloExporter.configure(EXPORT_QID).instance(instance, zookeepers)
+ .credentials(user, password).table(exportTable).save(fluoConfig);
- // Create config for export table.
- AccumuloExporter.Configuration exportTableCfg =
- new AccumuloExporter.Configuration(instance, zookeepers, user, password, exportTable);
-
- // Create config for export queue.
- ExportQueue.Options eqOpts = new ExportQueue.Options(EXPORT_QUEUE_ID, SimpleExporter.class,
- String.class, String.class, numMapBuckets).setExporterConfiguration(exportTableCfg);
-
- // Configure export queue. This will modify fluoConfig.
- ExportQueue.configure(fluoConfig, eqOpts);
+ // Set properties for export queue in Fluo app configuration
+ ExportQueue.configure(EXPORT_QID).keyType(String.class).valueType(String.class)
+ .buckets(119).save(fluoConfig);
// Initialize Fluo using fluoConfig
```
-3. Export queues can be retrieved in Fluo observers and objects can be added to them:
+3. In the applications `ObserverProvider`, register an observer that will process exports and write
+ them to Accumulo using [AccumuloExporter]. Also, register observers that add to the export
+ queue.
```java
- public class MyObserver extends AbstractObserver {
-
- ExportQueue<String, String> exportQ;
-
- @Override
- public void init(Context context) throws Exception {
- exportQ = ExportQueue.getInstance(EXPORT_QUEUE_ID, context.getAppConfiguration());
- }
+ public class MyObserverProvider implements ObserverProvider {
@Override
- public void process(TransactionBase tx, Bytes row, Column col) {
-
- // Read some data and do some work
-
- // Add results to export queue
- String key = // key that identifies export
- String value = // object to export
- export.add(tx, key, value);
+ public void provide(Registry obsRegistry, Context ctx) {
+ SimpleConfiguration appCfg = ctx.getAppConfiguration();
+
+ ExportQueue<String, String> expQ = ExportQueue.getInstance(EXPORT_QID, appCfg);
+
+ // Register observer that will processes entries on export queue and write them to the Accumulo
+ // table configured earlier. SimpleTranslator from step 1 is passed here, could have used a
+ // lambda instead.
+ expQ.registerObserver(obsRegistry,
+ new AccumuloExporter<>(EXPORT_QID, appCfg, new SimpleTranslator()));
+
+ // An example observer created using a lambda that adds to the export queue.
+ obsRegistry.register(OBS_COL, WEAK, (tx,row,col) -> {
+ // Read some data and do some work
+
+ // Add results to export queue
+ String key = // key that identifies export
+ String value = // object to export
+ expQ.add(tx, key, value);
+ });
}
+ }
```
## Other use cases
-[AccumuloReplicator] is a specialized [AccumuloExporter] that replicates a Fluo table to Accumulo.
+The `getTranslator()` method in [AccumuloReplicator] creates a specialized [AccumuloTranslator] for replicating a Fluo table to Accumulo.
[1]: export-queue.md
-[AccumuloExporter]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
+[Exporter]: ../modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java
+[AccumuloExporter]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloExporter.java
+[AccumuloTranslator]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloTranslator.java
[AccumuloReplicator]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/docs/cfm.md
----------------------------------------------------------------------
diff --git a/docs/cfm.md b/docs/cfm.md
deleted file mode 100644
index 85b05b0..0000000
--- a/docs/cfm.md
+++ /dev/null
@@ -1,257 +0,0 @@
-<!--
-Licensed to the Apache Software Foundation (ASF) under one or more
-contributor license agreements. See the NOTICE file distributed with
-this work for additional information regarding copyright ownership.
-The ASF licenses this file to You under the Apache License, Version 2.0
-(the "License"); you may not use this file except in compliance with
-the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
--->
-# Collision Free Map Recipe
-
-## Background
-
-When many transactions are trying to modify the same keys, collisions will occur.
-These collisions will cause the transactions to fail and throughput to nose
-dive. For example consider the [phrasecount] example. In this example many
-transactions are processing documents as input. Each transaction counts the
-phrases in its document and then tries to update global phrase counts. With
-each transaction attempting to update many phrase counts, the probability of
-two transactions colliding is very high.
-
-## Solution
-
-This recipe provides a reusable solution for the problem of many transactions
-updating many keys while avoiding collisions. As an added bonus, this recipe
-also organizes updates into batches for efficiency in order to improve
-throughput.
-
-The central idea behind this recipe is that updates to a key are queued up to
-be processed by another transaction triggered by weak notifications. In the
-phrase count example transactions processing documents would queue updates,
-but would not actually update the counts. Below is an example of how
-transactions would compute phrasecounts using this recipe.
-
- * TX1 queues `+1` update for phrase `we want lambdas now`
- * TX2 queues `+1` update for phrase `we want lambdas now`
- * TX3 reads the updates and current value for the phrase `we want lambdas now`. There is no current value and the updates sum to 2, so a new value of 2 is written.
- * TX4 queues `+2` update for phrase `we want lambdas now`
- * TX5 queues `-1` update for phrase `we want lambdas now`
- * TX6 reads the updates and current value for the phrase `we want lambdas now`. The current value is 2 and the updates sum to 1, so a new value of 3 is written.
-
-Transactions processing updates have the ability to make additional updates.
-For example in addition to updating the current value for a phrase, the new
-value could also be placed on an export queue to update an external database.
-
-### Buckets
-
-A simple implementation of this recipe would be to have an update queue for
-each key. However the implementation does something slightly more complex.
-Each update queue is in a bucket and transactions that process updates, process
-all of the updates in a bucket. This allows more efficient processing of
-updates for the following reasons :
-
- * When updates are queued, notifications are made per bucket(instead of per a key).
- * The transaction doing the update can scan the entire bucket reading updates, this avoids a seek for each key being updated.
- * Also the transaction can request a batch lookup to get the current value of all the keys being updated.
- * Any additional actions taken on update (like adding something to an export queue) can also be batched.
- * Data is organized to make reading exiting values for keys in a bucket more efficient.
-
-Which bucket a key goes to is decided using hash and modulus so that multiple
-updates for the same key always go to the same bucket.
-
-The initial number of tablets to create when applying table optimizations can be
-controlled by setting the buckets per tablet option when configuring a Collision
-Free Map. For example if you have 20 tablet servers and 1000 buckets and want
-2 tablets per tserver initially then set buckets per tablet to 1000/(2*20)=25.
-
-## Example Use
-
-The following code snippets show how to setup and use this recipe for
-wordcount. The first step in using this recipe is to configure it before
-initializing Fluo. When initializing an ID will need to be provided. This ID
-is used in two ways. First, the ID is used as a row prefix in the table.
-Therefore nothing else should use that row range in the table. Second, the ID
-is used in generating configuration keys associated with the instance of the
-Collision Free Map.
-
-The following snippet shows how to setup a collision free map.
-
-```java
- FluoConfiguration fluoConfig = ...;
-
- int numBuckets = 119;
-
- WordCountMap.configure(fluoConfig, 119);
-
- //initialize Fluo using fluoConfig
-
-```
-
-Assume the following observer is triggered when a documents contents are
-updated. It examines new and old document content and determines changes in
-word counts. These changes are pushed to a collision free map.
-
-```java
-public class DocumentObserver extends TypedObserver {
-
- CollisionFreeMap<String, Long> wcm;
-
- @Override
- public void init(Context context) throws Exception {
- wcm = CollisionFreeMap.getInstance(WordCountMap.ID, context.getAppConfiguration());
- }
-
- @Override
- public ObservedColumn getObservedColumn() {
- return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
- }
-
- @Override
- public void process(TypedTransactionBase tx, Bytes row, Column col) {
- String newContent = tx.get().row(row).col(col).toString();
- String currentContent = tx.get().row(row).fam("content").qual("current").toString("");
-
- Map<String, Long> newWordCounts = getWordCounts(newContent);
- Map<String, Long> currentWordCounts = getWordCounts(currentContent);
-
- //determine changes in word counts between old and new document content
- Map<String, Long> changes = calculateChanges(newWordCounts, currentWordCounts);
-
- //queue updates to word counts for processing by other transactions
- wcm.update(tx, changes);
-
- //update the current content and delete the new content
- tx.mutate().row(row).fam("content").qual("current").set(newContent);
- tx.mutate().row(row).col(col).delete();
- }
-
- private static Map<String, Long> getWordCounts(String doc) {
- //TODO extract words from doc
- }
-
- private static Map<String, Long> calculateChanges(Map<String, Long> newCounts,
- Map<String, Long> currCounts) {
- Map<String, Long> changes = new HashMap<>();
-
- // guava Maps class
- MapDifference<String, Long> diffs = Maps.difference(currCounts, newCounts);
-
- // compute the diffs for words that changed
- changes.putAll(Maps.transformValues(diffs.entriesDiffering(),
- vDiff -> vDiff.rightValue() - vDiff.leftValue()));
-
- // add all new words
- changes.putAll(diffs.entriesOnlyOnRight());
-
- // subtract all words no longer present
- changes.putAll(Maps.transformValues(diffs.entriesOnlyOnLeft(), l -> l * -1));
-
- return changes;
- }
-
-}
-```
-
-Each collision free map has two extension points, a combiner and an update
-observer. These two extension points are defined below as `WordCountCombiner`
-and `WordCountObserver`. The collision free map configures a Fluo observer that
-will process queued updates. When processing these queued updates the two
-extension points are called. In this example `WordCountCombiner` is called to
-combine updates that were queued by `DocumentObserver`. The collision free map
-will process a batch of keys, calling the combiner for each key. When finished
-processing a batch, it will call the update observer `WordCountObserver`.
-
-An update observer can do additional processing when a batch of key values are
-updated. In `WordCountObserver`, updates are queued for export to an external
-database. The export is given the new and old value allowing it to delete the
-old value if needed.
-
-```java
-/**
- * This class exists to provide a single place to put all code related to the
- * word count map.
- */
-public class WordCountMap {
-
- public static final String ID = "wc";
-
- /**
- * A helper method for configuring the word count map.
- *
- * @param numTablets the desired number of tablets to create when applying table optimizations
- */
- public static void configure(FluoConfiguration fluoConfig, int numBuckets, int numTablets) {
- Options cfmOpts =
- new Options(ID, WordCountCombiner.class, WordCountObserver.class, String.class, Long.class, numBuckets)
- .setBucketsPerTablet(numBuckets/numTablets);
- CollisionFreeMap.configure(fluoConfig, cfmOpts);
- }
-
- public static class WordCountCombiner implements Combiner<String, Long> {
- @Override
- public Optional<Long> combine(String key, Iterator<Long> updates) {
- long sum = 0L;
-
- while (updates.hasNext()) {
- sum += updates.next();
- }
-
- if (sum == 0) {
- //returning absent will cause the collision free map to delte the current key
- return Optional.absent();
- } else {
- return Optional.of(sum);
- }
- }
- }
-
- public static class WordCountObserver extends UpdateObserver<String, Long> {
-
- private ExportQueue<String, MyDatabaseExport> exportQ;
-
- @Override
- public void init(String mapId, Context observerContext) throws Exception {
- exportQ = ExportQueue.getInstance(MyExportQ.ID, observerContext.getAppConfiguration());
- }
-
- @Override
- public void updatingValues(TransactionBase tx, Iterator<Update<String, Long>> updates) {
- while (updates.hasNext()) {
- Update<String, Long> update = updates.next();
-
- String word = update.getKey();
- Optional<Long> oldVal = update.getOldValue();
- Optional<Long> newVal = update.getNewValue();
-
- //queue an export to let an external database know the word count has changed
- exportQ.add(word, new MyDatabaseExport(oldVal, newVal));
- }
- }
- }
-}
-```
-
-## Guarantees
-
-This recipe makes two important guarantees about updates for a key when it
-calls `updatingValues()` on an `UpdateObserver`.
-
- * The new value reported for an update will be derived from combining all
- updates that were committed before the transaction thats processing updates
- started. The implementation may have to make multiple passes over queued
- updates to achieve this. In the situation where TX1 queues a `+1` and later
- TX2 queues a `-1` for the same key, there is no need to worry about only seeing
- the `-1` processed. A transaction that started processing updates after TX2
- committed would process both.
- * The old value will always be what was reported as the new value in the
- previous transaction that called `updatingValues()`.
-
-[phrasecount]: https://github.com/fluo-io/phrasecount
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/docs/combine-queue.md
----------------------------------------------------------------------
diff --git a/docs/combine-queue.md b/docs/combine-queue.md
new file mode 100644
index 0000000..b3c1b13
--- /dev/null
+++ b/docs/combine-queue.md
@@ -0,0 +1,224 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+# Combine Queue Recipe
+
+## Background
+
+When many transactions try to modify the same keys, collisions will occur. Too many collisions
+cause transactions to fail and throughput to nose dive. For example, consider [phrasecount]
+which has many transactions processing documents. Each transaction counts the phrases in a document
+and then updates global phrase counts. Since transaction attempts to update many phrases
+, the probability of collisions is high.
+
+## Solution
+
+The [combine queue recipe][CombineQueue] provides a reusable solution for updating many keys while
+avoiding collisions. The recipe also organizes updates into batches in order to improve throughput.
+
+This recipes queues updates to keys for other transactions to process. In the phrase count example
+transactions processing documents queue updates, but do not actually update the counts. Below is an
+example of computing phrasecounts using this recipe.
+
+ * TX1 queues `+1` update for phrase `we want lambdas now`
+ * TX2 queues `+1` update for phrase `we want lambdas now`
+ * TX3 reads the updates and current value for the phrase `we want lambdas now`. There is no current value and the updates sum to 2, so a new value of 2 is written.
+ * TX4 queues `+2` update for phrase `we want lambdas now`
+ * TX5 queues `-1` update for phrase `we want lambdas now`
+ * TX6 reads the updates and current value for the phrase `we want lambdas now`. The current value is 2 and the updates sum to 1, so a new value of 3 is written.
+
+Transactions processing updates have the ability to make additional updates.
+For example in addition to updating the current value for a phrase, the new
+value could also be placed on an export queue to update an external database.
+
+### Buckets
+
+A simple implementation of this recipe would have an update queue for each key. However the
+implementation is slightly more complex. Each update queue is in a bucket and transactions process
+all of the updates in a bucket. This allows more efficient processing of updates for the following
+reasons :
+
+ * When updates are queued, notifications are made per bucket(instead of per a key).
+ * The transaction doing the update can scan the entire bucket reading updates, this avoids a seek for each key being updated.
+ * Also the transaction can request a batch lookup to get the current value of all the keys being updated.
+ * Any additional actions taken on update (like adding something to an export queue) can also be batched.
+ * Data is organized to make reading exiting values for keys in a bucket more efficient.
+
+Which bucket a key goes to is decided using hash and modulus so that multiple updates for a key go
+to the same bucket.
+
+The initial number of tablets to create when applying table optimizations can be controlled by
+setting the buckets per tablet option when configuring a Combine Queue. For example if you
+have 20 tablet servers and 1000 buckets and want 2 tablets per tserver initially then set buckets
+per tablet to 1000/(2*20)=25.
+
+## Example Use
+
+The following code snippets show how to use this recipe for wordcount. The first step is to
+configure it before initializing Fluo. When initializing an ID is needed. This ID is used in two
+ways. First, the ID is used as a row prefix in the table. Therefore nothing else should use that
+row range in the table. Second, the ID is used in generating configuration keys.
+
+The following snippet shows how to configure a combine queue.
+
+```java
+ FluoConfiguration fluoConfig = ...;
+
+ // Set application properties for the combine queue. These properties are read later by
+ // the observers running on each worker.
+ CombineQueue.configure(WcObserverProvider.ID)
+ .keyType(String.class).valueType(Long.class).buckets(119).save(fluoConfig);
+
+ fluoConfig.setObserverProvider(WcObserverProvider.class);
+
+ // initialize Fluo using fluoConfig
+```
+
+Assume the following observer is triggered when a documents is updated. It examines new
+and old document content and determines changes in word counts. These changes are pushed to a
+combine queue.
+
+```java
+public class DocumentObserver implements StringObserver {
+ // word count combine queue
+ private CombineQueue<String, Long> wccq;
+
+ public static final Column NEW_COL = new Column("content", "new");
+ public static final Column CUR_COL = new Column("content", "current");
+
+ public DocumentObserver(CombineQueue<String, Long> wccq) {
+ this.wccq = wccq;
+ }
+
+ @Override
+ public void process(TransactionBase tx, String row, Column col) {
+
+ Preconditions.checkArgument(col.equals(NEW_COL));
+
+ String newContent = tx.gets(row, NEW_COL);
+ String currentContent = tx.gets(row, CUR_COL, "");
+
+ Map<String, Long> newWordCounts = getWordCounts(newContent);
+ Map<String, Long> currentWordCounts = getWordCounts(currentContent);
+
+ // determine changes in word counts between old and new document content
+ Map<String, Long> changes = calculateChanges(newWordCounts, currentWordCounts);
+
+ // queue updates to word counts for processing by other transactions
+ wccq.add(tx, changes);
+
+ // update the current content and delete the new content
+ tx.set(row, CUR_COL, newContent);
+ tx.delete(row, NEW_COL);
+ }
+
+ private static Map<String, Long> getWordCounts(String doc) {
+ // TODO extract words from doc
+ }
+
+ private static Map<String, Long> calculateChanges(Map<String, Long> newCounts,
+ Map<String, Long> currCounts) {
+ Map<String, Long> changes = new HashMap<>();
+
+ // guava Maps class
+ MapDifference<String, Long> diffs = Maps.difference(currCounts, newCounts);
+
+ // compute the diffs for words that changed
+ changes.putAll(Maps.transformValues(diffs.entriesDiffering(),
+ vDiff -> vDiff.rightValue() - vDiff.leftValue()));
+
+ // add all new words
+ changes.putAll(diffs.entriesOnlyOnRight());
+
+ // subtract all words no longer present
+ changes.putAll(Maps.transformValues(diffs.entriesOnlyOnLeft(), l -> l * -1));
+
+ return changes;
+ }
+}
+
+
+```
+
+Each combine queue has two extension points, a [combiner][Combiner] and a [change
+observer][ChangeObserver]. The combine queue configures a Fluo observer to process queued
+updates. When processing updates the two extension points are called. The code below shows
+how to use these extension points.
+
+A change observer can do additional processing when a batch of key values are updated. Below
+updates are queued for export to an external database. The export is given the new and old value
+allowing it to delete the old value if needed.
+
+```java
+public class WcObserverProvider implements ObserverProvider {
+
+ public static final String ID = "wc";
+
+ @Override
+ public void provide(Registry obsRegistry, Context ctx) {
+
+ ExportQueue<String, MyDatabaseExport> exportQ = createExportQueue(ctx);
+
+ // Create a combine queue for computing word counts.
+ CombineQueue<String, Long> wcMap = CombineQueue.getInstance(ID, ctx.getAppConfiguration());
+
+ // Register observer that updates the Combine Queue
+ obsRegistry.register(DocumentObserver.NEW_COL, STRONG, new DocumentObserver(wcMap));
+
+ // Used to join new and existing values for a key. The lambda sums all values and returns
+ // Optional.empty() when the sum is zero. Returning Optional.empty() causes the key/value to be
+ // deleted. Could have used the built in SummingCombiner.
+ Combiner<String, Long> combiner = input -> input.stream().reduce(Long::sum).filter(l -> l != 0);
+
+ // Called when the value of a key changes. The lambda exports these changes to an external
+ // database. Make sure to read ChangeObserver's javadoc.
+ ChangeObserver<String, Long> changeObs = (tx, changes) -> {
+ for (Change<String, Long> update : changes) {
+ String word = update.getKey();
+ Optional<Long> oldVal = update.getOldValue();
+ Optional<Long> newVal = update.getNewValue();
+
+ // Queue an export to let an external database know the word count has changed.
+ exportQ.add(tx, word, new MyDatabaseExport(oldVal, newVal));
+ }
+ };
+
+ // Register observer that handles updates to the CombineQueue. This observer will use the
+ // combiner and valueObserver.
+ wcMap.registerObserver(obsRegistry, combiner, changeObs);
+ }
+}
+```
+
+## Guarantees
+
+This recipe makes two important guarantees about updates for a key when it
+calls `process()` on a [ChangeObserver].
+
+ * The new value reported for an update will be derived from combining all
+ updates that were committed before the transaction thats processing updates
+ started. The implementation may have to make multiple passes over queued
+ updates to achieve this. In the situation where TX1 queues a `+1` and later
+ TX2 queues a `-1` for the same key, there is no need to worry about only seeing
+ the `-1` processed. A transaction that started processing updates after TX2
+ committed would process both.
+ * The old value will always be what was reported as the new value in the
+ previous transaction that called `ChangeObserver.process()`.
+
+[phrasecount]: https://github.com/fluo-io/phrasecount
+[CombineQueue]: /modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueue.java
+[ChangeObserver]: /modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeObserver.java
+[Combiner]: /modules/core/src/main/java/org/apache/fluo/recipes/core/combine/Combiner.java
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/docs/export-queue.md
----------------------------------------------------------------------
diff --git a/docs/export-queue.md b/docs/export-queue.md
index 7ad931f..11e8d17 100644
--- a/docs/export-queue.md
+++ b/docs/export-queue.md
@@ -18,22 +18,20 @@ limitations under the License.
## Background
-Fluo is not suited for servicing low latency queries for two reasons. The
-first reason is that the implementation of transactions are designed for
-throughput. To get throughput, transactions recover lazily from failures and
-may wait on another transaction that is writing. Both of these design decisions
-can lead to delays for an individual transaction, but do not negatively impact
-throughput. The second reason is that Fluo observers executing transactions
-will likely cause a large number of random accesses. This could lead to high
-response time variability for an individual random access. This variability
-would not impede throughput but would impede the goal of latency.
+Fluo is not suited for servicing low latency queries for two reasons. First, the implementation of
+transactions are designed for throughput. To get throughput, transactions recover lazily from
+failures and may wait on other transactions. Both of these design decisions can
+lead to delays of individual transactions, but do not negatively impact throughput. The second
+reason is that Fluo observers executing transactions will likely cause a large number of random
+accesses. This could lead to high response time variability for an individual random access. This
+variability would not impede throughput but would impede the goal of low latency.
One way to make data transformed by Fluo available for low latency queries is
-to export that data to another system. For example Fluo could be running
+to export that data to another system. For example Fluo could run on
cluster A, continually transforming a large data set, and exporting data to
Accumulo tables on cluster B. The tables on cluster B would service user
queries. Fluo Recipes has built in support for [exporting to Accumulo][3],
-however recipe could be used to export to systems other than Accumulo, like
+however this recipe can be used to export to systems other than Accumulo, like
Redis, Elasticsearch, MySQL, etc.
Exporting data from Fluo is easy to get wrong which is why this recipe exists.
@@ -41,29 +39,30 @@ To understand what can go wrong consider the following example observer
transaction.
```java
-public class MyObserver extends AbstractObserver {
+public class MyObserver implements StringObserver {
- private static final TYPEL = new TypeLayer(new StringEncoder());
+ static final Column UPDATE_COL = new Column("meta", "numUpdates");
+ static final Column COUNTER_COL = new Column("meta", "counter1");
//reperesents a Query system extrnal to Fluo that is updated by Fluo
QuerySystem querySystem;
@Override
- public void process(TransactionBase tx, Bytes row, Column col) {
+ public void process(TransactionBase tx, String row, Column col) {
- TypedTransactionBase ttx = TYPEL.wrap(tx);
- int oldCount = ttx.get().row(row).fam("meta").qual("counter1").toInteger(0);
- int numUpdates = ttx.get().row(row).fam("meta").qual("numUpdates").toInteger(0);
- int newCount = oldCount + numUpdates;
- ttx.mutate().row(row).fam("meta").qual("counter1").set(newCount);
- ttx.mutate().row(row).fam("meta").qual("numUpdates").set(0);
+ int oldCount = Integer.parseInt(tx.gets(row, COUNTER_COL, "0"));
+ int numUpdates = Integer.parseInt(tx.gets(row, UPDATE_COL, "0"));
+ int newCount = oldCount + numUpdates;
+
+ tx.set(row, COUNTER_COL, "" + newCount);
+ tx.delete(row, UPDATE_COL);
//Build an inverted index in the query system, based on count from the
//meta:counter1 column in fluo. Do this by creating rows for the
- //external query system based on the count.
+ //external query system based on the count.
String oldCountRow = String.format("%06d", oldCount);
String newCountRow = String.format("%06d", newCount);
-
+
//add a new entry to the inverted index
querySystem.insertRow(newCountRow, row);
//remove the old entry from the inverted index
@@ -94,9 +93,9 @@ system before the transaction is committed (Note : for observers, the
transaction is committed by the framework after `process(...)` is called). So
if the transaction fails, the next time it runs it could compute a completely
different value for `newCountRow` (and it would not delete what was written by the
-failed transaction).
+failed transaction).
-## Solution
+## Solution
The simple solution to the problem of exporting uncommitted data is to only
export committed data. There are multiple ways to accomplish this. This
@@ -113,32 +112,18 @@ There are three requirements for using this recipe :
* Must configure export queues before initializing a Fluo application.
* Transactions adding to an export queue must get an instance of the queue using its unique QID.
- * Must implement a class that extends [Exporter][1] in order to process exports.
-
-## Schema
-
-Each export queue stores its data in the Fluo table in a contiguous row range.
-This row range is defined by using the export queue id as a row prefix for all
-data in the export queue. So the row range defined by the export queue id
-should not be used by anything else.
-
-All data stored in an export queue is [transient](transient.md). When an export
-queue is configured, it will recommend split points using the [table
-optimization process](table-optimization.md). The number of splits generated
-by this process can be controlled by setting the number of buckets per tablet
-when configuring an export queue.
+ * Must create a class or lambda that implements [Exporter][1] in order to process exports.
## Example Use
-This example will show how to build an inverted index in an external
-query system using an export queue. The class below is simple POJO to hold the
-count update, this will be used as the value for the export queue.
+This example shows how to incrementally build an inverted index in an external query system using an
+export queue. The class below is simple POJO used as the value for the export queue.
```java
class CountUpdate {
public int oldCount;
public int newCount;
-
+
public CountUpdate(int oc, int nc) {
this.oldCount = oc;
this.newCount = nc;
@@ -146,104 +131,134 @@ class CountUpdate {
}
```
-The following code shows how to configure an export queue. This code will
-modify the FluoConfiguration object with options needed for the export queue.
+The following code shows how to configure an export queue. This code
+modifies the FluoConfiguration object with options needed for the export queue.
This FluoConfiguration object should be used to initialize the fluo
application.
```java
- FluoConfiguration fluoConfig = ...;
-
- //queue id "ici" means inverted count index, would probably use static final in real application
- String exportQueueID = "ici";
- Class<CountExporter> exporterType = CountExporter.class;
- Class<Bytes> keyType = Bytes.class;
- Class<CountUpdate> valueType = CountUpdate.class;
- int numBuckets = 1009;
- //the desired number of tablets to create when applying table optimizations
- int numTablets = 100;
-
- ExportQueue.Options eqOptions =
- new ExportQueue.Options(exportQueueId, exporterType, keyType, valueType, numBuckets)
- .setsetBucketsPerTablet(numBuckets/numTablets);
- ExportQueue.configure(fluoConfig, eqOptions);
-
- //initialize Fluo using fluoConfig
-```
+public class FluoApp {
-Below is updated version of the observer from above thats now using the export
-queue.
+ // export queue id "ici" means inverted count index
+ public static final String EQ_ID = "ici";
-```java
-public class MyObserver extends AbstractObserver {
+ static final Column UPDATE_COL = new Column("meta", "numUpdates");
+ static final Column COUNTER_COL = new Column("meta", "counter1");
- private static final TYPEL = new TypeLayer(new StringEncoder());
+ public static class AppObserverProvider implements ObserverProvider {
+ @Override
+ public void provide(Registry obsRegistry, Context ctx) {
+ ExportQueue<String, CountUpdate> expQ =
+ ExportQueue.getInstance(EQ_ID, ctx.getAppConfiguration());
- private ExportQueue<Bytes, CountUpdate> exportQueue;
+ // register observer that will queue data to export
+ obsRegistry.register(UPDATE_COL, STRONG, new MyObserver(expQ));
- @Override
- public void init(Context context) throws Exception {
- exportQueue = ExportQueue.getInstance("ici", context.getAppConfiguration());
+ // register observer that will export queued data
+ expQ.registerObserver(obsRegistry, new CountExporter());
}
+ }
- @Override
- public void process(TransactionBase tx, Bytes row, Column col) {
-
- TypedTransactionBase ttx = TYPEL.wrap(tx);
- int oldCount = ttx.get().row(row).fam("meta").qual("counter1").toInteger(0);
- int numUpdates = ttx.get().row(row).fam("meta").qual("numUpdates").toInteger(0);
- int newCount = oldCount + numUpdates;
- ttx.mutate().row(row).fam("meta").qual("counter1").set(newCount);
- ttx.mutate().row(row).fam("meta").qual("numUpdates").set(0);
-
- //Because the update to the export queue is part of the transaction,
- //either the update to meta:counter1 is made and an entry is added to
- //the export queue or neither happens.
- exportQueue.add(tx, row, new CountUpdate(oldCount, newCount));
- }
+ /**
+ * Call this method before initializing Fluo.
+ *
+ * @param fluoConfig the configuration object that will be used to initialize Fluo
+ */
+ public static void preInit(FluoConfiguration fluoConfig) {
+
+ // Set properties for export queue in Fluo app configuration
+ ExportQueue.configure(QUEUE_ID)
+ .keyType(String.class)
+ .valueType(CountUpdate.class)
+ .buckets(1009)
+ .bucketsPerTablet(10)
+ .save(getFluoConfiguration());
+
+ fluoConfig.setObserverProvider(AppObserverProvider.class);
+ }
}
```
-The observer setup for the export queue will call the `processExports()` method
-on the class below to process entries in the export queue. It possible the
-call to `processExports()` can fail part way through and/or be called multiple
-times. In the case of failures the exporter will eventually be called again
-with the exact same data. The possibility of the same export entry being
-processed on multiple computer at different times can cause exports to arrive
-out of order. The system receiving exports has to be resilient to data
-arriving out of order and multiple times. The purpose of the sequence number
-is to help systems receiving data from Fluo process out of order and redundant
-data.
+Below is updated version of the observer from above thats now using an export
+queue.
```java
- public class CountExporter extends Exporter<Bytes, CountUpdate> {
- //represents the external query system we want to update from Fluo
- QuerySystem querySystem;
+public class MyObserver implements StringObserver {
- @Override
- protected void processExports(Iterator<SequencedExport<Bytes, CountUpdate>> exportIterator) {
- BatchUpdater batchUpdater = querySystem.getBatchUpdater();
- while(exportIterator.hasNext()){
- SequencedExport<Bytes, CountUpdate> exportEntry = exportItertor.next();
- Bytes row = exportEntry.getKey();
- UpdateCount uc = exportEntry.getValue();
- long seqNum = exportEntry.getSequence();
+ private ExportQueue<String, CountUpdate> exportQueue;
- String oldCountRow = String.format("%06d", uc.oldCount);
- String newCountRow = String.format("%06d", uc.newCount);
+ public MyObserver(ExportQueue<String, CountUpdate> exportQueue) {
+ this.exportQueue = exportQueue;
+ }
- //add a new entry to the inverted index
- batchUpdater.insertRow(newCountRow, row, seqNum);
- //remove the old entry from the inverted index
- batchUpdater.deleteRow(oldCountRow, row, seqNum);
- }
+ @Override
+ public void process(TransactionBase tx, String row, Column col) {
+
+ int oldCount = Integer.parseInt(tx.gets(row, FluoApp.COUNTER_COL, "0"));
+ int numUpdates = Integer.parseInt(tx.gets(row, FluoApp.UPDATE_COL, "0"));
+ int newCount = oldCount + numUpdates;
+
+ tx.set(row, FluoApp.COUNTER_COL, "" + newCount);
+ tx.delete(row, FluoApp.UPDATE_COL);
+
+ // Because the update to the export queue is part of the transaction,
+ // either the update to meta:counter1 is made and an entry is added to
+ // the export queue or neither happens.
+ exportQueue.add(tx, row, new CountUpdate(oldCount, newCount));
+ }
+}
+```
- //flush all of the updates to the external query system
- batchUpdater.close();
+The export queue will call the `accept()` method on the class below to process entries queued for
+export. It is possible the call to `accept()` can fail part way through and/or be called multiple
+times. In the case of failures the export consumer will be called again with the same data.
+Its possible for the same export entry to be processed on multiple computers at different times.
+This can cause exports to arrive out of order. The purpose of the sequence number is to help
+systems receiving out of order and redundant data.
+
+```java
+public class CountExporter implements Exporter<String, CountUpdate> {
+ // represents the external query system we want to update from Fluo
+ QuerySystem querySystem;
+
+ @Override
+ public void export(Iterator<SequencedExport<String, CountUpdate>> exports) {
+ BatchUpdater batchUpdater = querySystem.getBatchUpdater();
+
+ while (exports.hasNext()) {
+ SequencedExport<String, CountUpdate> export = exports.next();
+ String row = export.getKey();
+ CountUpdate uc = export.getValue();
+ long seqNum = export.getSequence();
+
+ String oldCountRow = String.format("%06d", uc.oldCount);
+ String newCountRow = String.format("%06d", uc.newCount);
+
+ // add a new entry to the inverted index
+ batchUpdater.insertRow(newCountRow, row, seqNum);
+ // remove the old entry from the inverted index
+ batchUpdater.deleteRow(oldCountRow, row, seqNum);
}
+
+ // flush all of the updates to the external query system
+ batchUpdater.close();
}
+}
```
+## Schema
+
+Each export queue stores its data in the Fluo table in a contiguous row range.
+This row range is defined by using the export queue id as a row prefix for all
+data in the export queue. So the row range defined by the export queue id
+should not be used by anything else.
+
+All data stored in an export queue is [transient](transient.md). When an export
+queue is configured, it will recommend split points using the [table
+optimization process](table-optimization.md). The number of splits generated
+by this process can be controlled by setting the number of buckets per tablet
+when configuring an export queue.
+
## Concurrency
Additions to the export queue will never collide. If two transactions add the
@@ -285,7 +300,7 @@ cells (`rowB fam1:qual2` and `rowA fam1:qual2`). This approach makes it more
difficult to reason about export entries with the same key, because the
transactions adding those entries could have overlapped in time. This is an
example of write skew mentioned in the Percolater paper.
-
+
1. TH1 : key1 = ek(`row1`,`fam1:qual1`)
1. TH1 : val1 = ev(tx1.get(`row1`,`fam1:qual1`), tx1.get(`rowA`,`fam1:qual2`))
1. TH1 : exportQueueA.add(tx1, key1, val1)
@@ -295,7 +310,7 @@ example of write skew mentioned in the Percolater paper.
1. TH1 : tx1.set(`rowA`,`fam1:qual2`, val1)
1. TH2 : tx2.set(`rowB`,`fam1:qual2`, val2)
-[1]: ../modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
+[1]: ../modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java
[2]: https://en.wikipedia.org/wiki/Serializability
[3]: accumulo-export-queue.md
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
index 6a19362..b2bef84 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
@@ -15,18 +15,17 @@
package org.apache.fluo.recipes.accumulo.export;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.function.Consumer;
import org.apache.accumulo.core.data.Mutation;
+import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.data.RowColumn;
-import org.apache.fluo.recipes.core.export.ExportQueue;
+import org.apache.fluo.recipes.core.export.ExportQueue.Options;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloTranslator;
import org.apache.fluo.recipes.core.export.Exporter;
import org.apache.fluo.recipes.core.export.SequencedExport;
@@ -35,17 +34,23 @@ import org.apache.fluo.recipes.core.export.SequencedExport;
* to use this, see the project level documentation for exporting to Accumulo.
*
* @since 1.0.0
+ * @deprecated since 1.1.0, replaced by
+ * {@link org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter} and
+ * {@link AccumuloTranslator}
*/
+@Deprecated
public abstract class AccumuloExporter<K, V> extends Exporter<K, V> {
/**
* Use this to configure the Accumulo table where an AccumuloExporter's mutations will be written.
- * Create and pass to {@link ExportQueue.Options#setExporterConfiguration(SimpleConfiguration)}
+ * Create and pass to {@link Options#setExporterConfiguration(SimpleConfiguration)}
*
* @since 1.0.0
*/
public static class Configuration extends SimpleConfiguration {
+ private static final long serialVersionUID = 1L;
+
public Configuration(String instanceName, String zookeepers, String user, String password,
String table) {
super.setProperty("instanceName", instanceName);
@@ -56,28 +61,28 @@ public abstract class AccumuloExporter<K, V> extends Exporter<K, V> {
}
}
- private AccumuloWriter accumuloWriter;
+ private org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter<K, V> accumuloWriter;
@Override
public void init(Exporter.Context context) throws Exception {
- accumuloWriter = AccumuloWriter.getInstance(context.getExporterConfiguration());
+ SimpleConfiguration sc = context.getExporterConfiguration();
+ String instanceName = sc.getString("instanceName");
+ String zookeepers = sc.getString("zookeepers");
+ String user = sc.getString("user");
+ String password = sc.getString("password");
+ String table = sc.getString("table");
+
+ FluoConfiguration tmpFc = new FluoConfiguration();
+ org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter.configure("aecfgid")
+ .instance(instanceName, zookeepers).credentials(user, password).table(table).save(tmpFc);
+ accumuloWriter =
+ new org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter<K, V>("aecfgid",
+ tmpFc.getAppConfiguration(), this::translate);
}
@Override
protected void processExports(Iterator<SequencedExport<K, V>> exports) {
-
- ArrayList<Mutation> buffer = new ArrayList<>();
-
- Consumer<Mutation> consumer = m -> buffer.add(m);
-
- while (exports.hasNext()) {
- SequencedExport<K, V> export = exports.next();
- translate(export, consumer);
- }
-
- if (buffer.size() > 0) {
- accumuloWriter.write(buffer);
- }
+ accumuloWriter.export(exports);
}
/**
@@ -108,29 +113,11 @@ public abstract class AccumuloExporter<K, V> extends Exporter<K, V> {
* @param oldData Map containing old row/column data
* @param newData Map containing new row/column data
* @param seq Export sequence number
+ * @deprecated since 1.1.0 use
+ * {@link AccumuloTranslator#generateMutations(long, Map, Map, Consumer)}
*/
public static void generateMutations(long seq, Map<RowColumn, Bytes> oldData,
Map<RowColumn, Bytes> newData, Consumer<Mutation> consumer) {
- Map<Bytes, Mutation> mutationMap = new HashMap<>();
- for (Map.Entry<RowColumn, Bytes> entry : oldData.entrySet()) {
- RowColumn rc = entry.getKey();
- if (!newData.containsKey(rc)) {
- Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
- m.putDelete(rc.getColumn().getFamily().toArray(), rc.getColumn().getQualifier().toArray(),
- seq);
- }
- }
- for (Map.Entry<RowColumn, Bytes> entry : newData.entrySet()) {
- RowColumn rc = entry.getKey();
- Column col = rc.getColumn();
- Bytes newVal = entry.getValue();
- Bytes oldVal = oldData.get(rc);
- if (oldVal == null || !oldVal.equals(newVal)) {
- Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
- m.put(col.getFamily().toArray(), col.getQualifier().toArray(), seq, newVal.toArray());
- }
- }
-
- mutationMap.values().forEach(consumer);
+ AccumuloTranslator.generateMutations(seq, oldData, newData, consumer);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
index ed16fc7..c2b9ade 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
@@ -24,22 +24,35 @@ import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.api.data.Column;
+import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloTranslator;
import org.apache.fluo.recipes.core.export.SequencedExport;
import org.apache.fluo.recipes.core.transaction.LogEntry;
+import org.apache.fluo.recipes.core.transaction.RecordingTransaction;
import org.apache.fluo.recipes.core.transaction.TxLog;
/**
- * An {@link AccumuloExporter} that replicates data to Accumulo using a {@link TxLog}
+ * Supports replicating data to Accumulo using a {@link TxLog}. The method {@link #getTranslator()}
+ * can be used with {@link org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter} to
+ * export {@link TxLog} objects.
*/
+@SuppressWarnings("deprecation")
public class AccumuloReplicator extends AccumuloExporter<String, TxLog> {
+ /**
+ * @deprecated since 1.1.0 use
+ * {@link org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter} with
+ * {@link #getTranslator()} instead.
+ */
@Override
protected void translate(SequencedExport<String, TxLog> export, Consumer<Mutation> consumer) {
generateMutations(export.getSequence(), export.getValue(), consumer);
}
/**
- * Returns LogEntry filter for Accumulo replication
+ * Returns LogEntry filter for Accumulo replication.
+ *
+ * @see RecordingTransaction#wrap(org.apache.fluo.api.client.TransactionBase, Predicate)
*/
public static Predicate<LogEntry> getFilter() {
return le -> le.getOp().equals(LogEntry.Operation.DELETE)
@@ -47,6 +60,15 @@ public class AccumuloReplicator extends AccumuloExporter<String, TxLog> {
}
/**
+ * @return A translator from TxLog to Mutations
+ * @since 1.1.0
+ */
+ public static AccumuloTranslator<String, TxLog> getTranslator() {
+ return (export, consumer) -> generateMutations(export.getSequence(), export.getValue(),
+ consumer);
+ }
+
+ /**
* Generates Accumulo mutations from a Transaction log. Used to Replicate Fluo table to Accumulo.
*
* @param txLog Transaction log
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
deleted file mode 100644
index 9016a2c..0000000
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License. You may obtain a
- * copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package org.apache.fluo.recipes.accumulo.export;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.BatchWriterConfig;
-import org.apache.accumulo.core.client.ClientConfiguration;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.client.ZooKeeperInstance;
-import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.fluo.api.config.SimpleConfiguration;
-
-/**
- * Writes mutations to Accumulo using a shared batch writer
- *
- * @since 1.0.0
- */
-class AccumuloWriter {
-
- private static class Mutations {
- List<Mutation> mutations;
- CountDownLatch cdl = new CountDownLatch(1);
-
- Mutations(Collection<Mutation> mutations) {
- this.mutations = new ArrayList<>(mutations);
- }
- }
-
- private static class ExportTask implements Runnable {
-
- private BatchWriter bw;
-
- ExportTask(String instanceName, String zookeepers, String user, String password, String table)
- throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
- ZooKeeperInstance zki =
- new ZooKeeperInstance(new ClientConfiguration().withInstance(instanceName).withZkHosts(
- zookeepers));
-
- // TODO need to close batch writer
- Connector conn = zki.getConnector(user, new PasswordToken(password));
- try {
- bw = conn.createBatchWriter(table, new BatchWriterConfig());
- } catch (TableNotFoundException tnfe) {
- try {
- conn.tableOperations().create(table);
- } catch (TableExistsException e) {
- // nothing to do
- }
-
- bw = conn.createBatchWriter(table, new BatchWriterConfig());
- }
- }
-
- @Override
- public void run() {
-
- ArrayList<AccumuloWriter.Mutations> exports = new ArrayList<>();
-
- while (true) {
- try {
- exports.clear();
-
- // gather export from all threads that have placed an item on the queue
- exports.add(exportQueue.take());
- exportQueue.drainTo(exports);
-
- for (AccumuloWriter.Mutations ml : exports) {
- bw.addMutations(ml.mutations);
- }
-
- bw.flush();
-
- // notify all threads waiting after flushing
- for (AccumuloWriter.Mutations ml : exports) {
- ml.cdl.countDown();
- }
-
- } catch (InterruptedException | MutationsRejectedException e) {
- throw new RuntimeException(e);
- }
- }
- }
-
- }
-
- private static LinkedBlockingQueue<AccumuloWriter.Mutations> exportQueue = null;
-
- private AccumuloWriter(String instanceName, String zookeepers, String user, String password,
- String table) {
-
- // TODO: fix this write to static and remove findbugs max rank override in pom.xml
- exportQueue = new LinkedBlockingQueue<>(10000);
-
- try {
- Thread queueProcessingTask =
- new Thread(new ExportTask(instanceName, zookeepers, user, password, table));
- queueProcessingTask.setDaemon(true);
- queueProcessingTask.start();
- } catch (Exception e) {
- throw new IllegalStateException(e);
- }
- }
-
- private static Map<String, AccumuloWriter> exporters = new HashMap<>();
-
-
- static AccumuloWriter getInstance(SimpleConfiguration sc) {
- String instanceName = sc.getString("instanceName");
- String zookeepers = sc.getString("zookeepers");
- String user = sc.getString("user");
- String password = sc.getString("password");
- String table = sc.getString("table");
- return getInstance(instanceName, zookeepers, user, password, table);
- }
-
- static synchronized AccumuloWriter getInstance(String instanceName, String zookeepers,
- String user, String password, String table) {
-
- String key =
- instanceName + ":" + zookeepers + ":" + user + ":" + password.hashCode() + ":" + table;
-
- AccumuloWriter ret = exporters.get(key);
-
- if (ret == null) {
- ret = new AccumuloWriter(instanceName, zookeepers, user, password, table);
- exporters.put(key, ret);
- }
-
- return ret;
- }
-
- void write(Collection<Mutation> mutations) {
- AccumuloWriter.Mutations work = new Mutations(mutations);
- exportQueue.add(work);
- try {
- work.cdl.await();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloExporter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloExporter.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloExporter.java
new file mode 100644
index 0000000..039dc55
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloExporter.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.accumulo.export.function;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.function.Consumer;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.recipes.core.export.ExportQueue;
+import org.apache.fluo.recipes.core.export.SequencedExport;
+import org.apache.fluo.recipes.core.export.function.Exporter;
+import org.apache.fluo.api.observer.ObserverProvider.Registry;
+
+/**
+ * An Accumulo-specific {@link Exporter} that writes mutations to Accumulo. For an overview of how
+ * to use this, see the project level documentation for exporting to Accumulo.
+ *
+ * @see ExportQueue#registerObserver(Registry, Exporter)
+ * @since 1.1.0
+ */
+public class AccumuloExporter<K, V> implements Exporter<K, V> {
+
+ private AccumuloTranslator<K, V> translator;
+ private AccumuloWriter writer;
+
+ /**
+ * Part of a fluent configuration API.
+ *
+ * @since 1.1.0
+ */
+ public static interface InstanceArgs {
+ CredentialArgs instance(String instanceName, String zookeepers);
+ }
+
+ /**
+ * Part of a fluent configuration API.
+ *
+ * @since 1.1.0
+ */
+ public static interface CredentialArgs {
+ TableArgs credentials(String user, String password);
+ }
+
+ /**
+ * Part of a fluent configuration API.
+ *
+ * @since 1.1.0
+ */
+ public static interface TableArgs {
+ Options table(String tableName);
+ }
+
+ /**
+ * Part of a fluent configuration API.
+ *
+ * @since 1.1.0
+ */
+ public static interface Options {
+ void save(FluoConfiguration fluoConf);
+ }
+
+ public static InstanceArgs configure(String configId) {
+ return new AeFluentConfigurator(configId);
+ }
+
+ public AccumuloExporter(String configId, SimpleConfiguration appConfig,
+ AccumuloTranslator<K, V> translator) {
+ AeFluentConfigurator cfg = AeFluentConfigurator.load(configId, appConfig);
+ this.writer =
+ AccumuloWriter.getInstance(cfg.getInstance(), cfg.getZookeepers(), cfg.getUser(),
+ cfg.getPassword(), cfg.getTable());
+ this.translator = translator;
+ }
+
+ @Override
+ public void export(Iterator<SequencedExport<K, V>> t) {
+ ArrayList<Mutation> buffer = new ArrayList<>();
+ Consumer<Mutation> consumer = m -> buffer.add(m);
+
+ while (t.hasNext()) {
+ translator.translate(t.next(), consumer);
+ }
+
+ if (buffer.size() > 0) {
+ writer.write(buffer);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloTranslator.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloTranslator.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloTranslator.java
new file mode 100644
index 0000000..c6b48b2
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloTranslator.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.accumulo.export.function;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.recipes.core.export.SequencedExport;
+
+/**
+ * This interface is used by {@link AccumuloExporter} to translated exports into Accumulo mutations.
+ *
+ * @see AccumuloExporter
+ * @since 1.1.0
+ */
+@FunctionalInterface
+public interface AccumuloTranslator<K, V> {
+ /**
+ * This function should convert the export to zero or more mutations, passing the mutations to the
+ * consumer.
+ */
+ void translate(SequencedExport<K, V> export, Consumer<Mutation> mutationWriter);
+
+ /**
+ * Generates Accumulo mutations by comparing the differences between a RowColumn/Bytes map that is
+ * generated for old and new data and represents how the data should exist in Accumulo. When
+ * comparing each row/column/value (RCV) of old and new data, mutations are generated using the
+ * following rules:
+ * <ul>
+ * <li>If old and new data have the same RCV, nothing is done.
+ * <li>If old and new data have same row/column but different values, an update mutation is
+ * created for the row/column.
+ * <li>If old data has a row/column that is not in the new data, a delete mutation is generated.
+ * <li>If new data has a row/column that is not in the old data, an insert mutation is generated.
+ * <li>Only one mutation is generated per row.
+ * <li>The export sequence number is used for the timestamp in the mutation.
+ * </ul>
+ *
+ * @param consumer generated mutations will be output to this consumer
+ * @param oldData Map containing old row/column data
+ * @param newData Map containing new row/column data
+ * @param seq Export sequence number
+ */
+ public static void generateMutations(long seq, Map<RowColumn, Bytes> oldData,
+ Map<RowColumn, Bytes> newData, Consumer<Mutation> consumer) {
+ Map<Bytes, Mutation> mutationMap = new HashMap<>();
+ for (Map.Entry<RowColumn, Bytes> entry : oldData.entrySet()) {
+ RowColumn rc = entry.getKey();
+ if (!newData.containsKey(rc)) {
+ Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
+ m.putDelete(rc.getColumn().getFamily().toArray(), rc.getColumn().getQualifier().toArray(),
+ seq);
+ }
+ }
+ for (Map.Entry<RowColumn, Bytes> entry : newData.entrySet()) {
+ RowColumn rc = entry.getKey();
+ Column col = rc.getColumn();
+ Bytes newVal = entry.getValue();
+ Bytes oldVal = oldData.get(rc);
+ if (oldVal == null || !oldVal.equals(newVal)) {
+ Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
+ m.put(col.getFamily().toArray(), col.getQualifier().toArray(), seq, newVal.toArray());
+ }
+ }
+
+ mutationMap.values().forEach(consumer);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloWriter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloWriter.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloWriter.java
new file mode 100644
index 0000000..0c7fc85
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AccumuloWriter.java
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.accumulo.export.function;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.ClientConfiguration;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Mutation;
+
+/**
+ * Writes mutations to Accumulo using a shared batch writer
+ *
+ * @since 1.0.0
+ */
+// intentionally package private
+class AccumuloWriter {
+
+ private static class Mutations {
+ List<Mutation> mutations;
+ CountDownLatch cdl = new CountDownLatch(1);
+
+ Mutations(Collection<Mutation> mutations) {
+ this.mutations = new ArrayList<>(mutations);
+ }
+ }
+
+ private static class ExportTask implements Runnable {
+
+ private BatchWriter bw;
+
+ ExportTask(String instanceName, String zookeepers, String user, String password, String table)
+ throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
+ ZooKeeperInstance zki =
+ new ZooKeeperInstance(new ClientConfiguration().withInstance(instanceName).withZkHosts(
+ zookeepers));
+
+ // TODO need to close batch writer
+ Connector conn = zki.getConnector(user, new PasswordToken(password));
+ try {
+ bw = conn.createBatchWriter(table, new BatchWriterConfig());
+ } catch (TableNotFoundException tnfe) {
+ try {
+ conn.tableOperations().create(table);
+ } catch (TableExistsException e) {
+ // nothing to do
+ }
+
+ bw = conn.createBatchWriter(table, new BatchWriterConfig());
+ }
+ }
+
+ @Override
+ public void run() {
+
+ ArrayList<AccumuloWriter.Mutations> exports = new ArrayList<>();
+
+ while (true) {
+ try {
+ exports.clear();
+
+ // gather export from all threads that have placed an item on the queue
+ exports.add(exportQueue.take());
+ exportQueue.drainTo(exports);
+
+ for (AccumuloWriter.Mutations ml : exports) {
+ bw.addMutations(ml.mutations);
+ }
+
+ bw.flush();
+
+ // notify all threads waiting after flushing
+ for (AccumuloWriter.Mutations ml : exports) {
+ ml.cdl.countDown();
+ }
+
+ } catch (InterruptedException | MutationsRejectedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ }
+
+ private static LinkedBlockingQueue<AccumuloWriter.Mutations> exportQueue = null;
+
+ private AccumuloWriter(String instanceName, String zookeepers, String user, String password,
+ String table) {
+
+ // TODO: fix this write to static and remove findbugs max rank override in pom.xml
+ exportQueue = new LinkedBlockingQueue<>(10000);
+
+ try {
+ Thread queueProcessingTask =
+ new Thread(new ExportTask(instanceName, zookeepers, user, password, table));
+ queueProcessingTask.setDaemon(true);
+ queueProcessingTask.start();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static Map<String, AccumuloWriter> exporters = new HashMap<>();
+
+ static synchronized AccumuloWriter getInstance(String instanceName, String zookeepers,
+ String user, String password, String table) {
+
+ String key =
+ instanceName + ":" + zookeepers + ":" + user + ":" + password.hashCode() + ":" + table;
+
+ AccumuloWriter ret = exporters.get(key);
+
+ if (ret == null) {
+ ret = new AccumuloWriter(instanceName, zookeepers, user, password, table);
+ exporters.put(key, ret);
+ }
+
+ return ret;
+ }
+
+ void write(Collection<Mutation> mutations) {
+ AccumuloWriter.Mutations work = new Mutations(mutations);
+ exportQueue.add(work);
+ try {
+ work.cdl.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AeFluentConfigurator.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AeFluentConfigurator.java b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AeFluentConfigurator.java
new file mode 100644
index 0000000..b146e3d
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/function/AeFluentConfigurator.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.accumulo.export.function;
+
+import java.util.Objects;
+
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter.CredentialArgs;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter.InstanceArgs;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter.Options;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloExporter.TableArgs;
+
+// Intentionally package private
+class AeFluentConfigurator implements InstanceArgs, CredentialArgs, TableArgs, Options {
+
+ private String id;
+ private String instance;
+ private String zookeepers;
+ private String user;
+ private String password;
+ private String table;
+
+ private static final String PREFIX = "recipes.accumulo.writer.";
+
+ AeFluentConfigurator(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public void save(FluoConfiguration fluoConf) {
+ SimpleConfiguration appConfig = fluoConf.getAppConfiguration();
+ // TODO Auto-generated method stub
+ appConfig.setProperty(PREFIX + id + ".instance", instance);
+ appConfig.setProperty(PREFIX + id + ".zookeepers", zookeepers);
+ appConfig.setProperty(PREFIX + id + ".user", user);
+ appConfig.setProperty(PREFIX + id + ".password", password);
+ appConfig.setProperty(PREFIX + id + ".table", table);
+ }
+
+ @Override
+ public Options table(String tableName) {
+ this.table = Objects.requireNonNull(tableName);
+ return this;
+ }
+
+ @Override
+ public TableArgs credentials(String user, String password) {
+ this.user = Objects.requireNonNull(user);
+ this.password = Objects.requireNonNull(password);
+ return this;
+ }
+
+ @Override
+ public CredentialArgs instance(String instanceName, String zookeepers) {
+ this.instance = Objects.requireNonNull(instanceName);
+ this.zookeepers = Objects.requireNonNull(zookeepers);
+ return this;
+ }
+
+ String getInstance() {
+ return instance;
+ }
+
+ String getZookeepers() {
+ return zookeepers;
+ }
+
+ String getUser() {
+ return user;
+ }
+
+ String getPassword() {
+ return password;
+ }
+
+ String getTable() {
+ return table;
+ }
+
+ public static AeFluentConfigurator load(String id, SimpleConfiguration config) {
+ AeFluentConfigurator aefc = new AeFluentConfigurator(id);
+ aefc.instance = config.getString(PREFIX + id + ".instance");
+ aefc.zookeepers = config.getString(PREFIX + id + ".zookeepers");
+ aefc.user = config.getString(PREFIX + id + ".user");
+ aefc.password = config.getString(PREFIX + id + ".password");
+ aefc.table = config.getString(PREFIX + id + ".table");
+
+ return aefc;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java b/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
deleted file mode 100644
index 53d2013..0000000
--- a/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance with the License. You may obtain a
- * copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-
-package org.apache.fluo.recipes.accumulo.export;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.function.Consumer;
-
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumn;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class AccumuloExportTest {
-
- public static Map<RowColumn, Bytes> genData(String key, Optional<String> val) {
- if (!val.isPresent()) {
- return Collections.emptyMap();
- }
- Map<RowColumn, Bytes> rcMap = new HashMap<>();
- String data = val.get();
- for (int i = 0; i < data.length(); i++) {
- char c = data.charAt(i);
- rcMap.put(new RowColumn("r:" + key, new Column("cf:" + c)), Bytes.of("v:" + c));
- }
- return rcMap;
- }
-
- public static void genMutations(String key, long seq, Optional<String> oldVal,
- Optional<String> newVal, Consumer<Mutation> consumer) {
- AccumuloExporter.generateMutations(seq, genData(key, oldVal), genData(key, newVal), consumer);
- }
-
- public static Mutation makePut(String key, String val, long seq) {
- Mutation m = new Mutation("r:" + key);
- addPut(m, key, val, seq);
- return m;
- }
-
- public static void addPut(Mutation m, String key, String val, long seq) {
- m.put("cf:" + val, "", seq, "v:" + val);
- }
-
- public static Mutation makeDel(String key, String val, long seq) {
- Mutation m = new Mutation("r:" + key);
- addDel(m, key, val, seq);
- return m;
- }
-
- public static void addDel(Mutation m, String key, String val, long seq) {
- m.putDelete("cf:" + val, "", seq);
- }
-
- @Test
- public void testDifferenceExport() {
- final Collection<Mutation> mutations = new ArrayList<>();
- Consumer<Mutation> consumer = m -> mutations.add(m);
-
- genMutations("k1", 1, Optional.empty(), Optional.of("a"), consumer);
- Assert.assertEquals(1, mutations.size());
- Assert.assertTrue(mutations.contains(makePut("k1", "a", 1)));
- mutations.clear();
-
- genMutations("k2", 2, Optional.of("ab"), Optional.of("ab"), consumer);
- Assert.assertEquals(0, mutations.size());
- mutations.clear();
-
- genMutations("k2", 2, Optional.of("b"), Optional.of("ab"), consumer);
- Assert.assertEquals(1, mutations.size());
- Assert.assertTrue(mutations.contains(makePut("k2", "a", 2)));
- mutations.clear();
-
- genMutations("k3", 3, Optional.of("c"), Optional.of("d"), consumer);
- Assert.assertEquals(1, mutations.size());
- Mutation m = makeDel("k3", "c", 3);
- addPut(m, "k3", "d", 3);
- Assert.assertTrue(mutations.contains(m));
- mutations.clear();
-
- genMutations("k4", 4, Optional.of("e"), Optional.empty(), consumer);
- Assert.assertEquals(1, mutations.size());
- Assert.assertTrue(mutations.contains(makeDel("k4", "e", 4)));
- mutations.clear();
-
- genMutations("k5", 5, Optional.of("ef"), Optional.of("fg"), consumer);
- Assert.assertEquals(1, mutations.size());
- m = makeDel("k5", "e", 5);
- addPut(m, "k5", "g", 5);
- Assert.assertTrue(mutations.contains(m));
- mutations.clear();
- }
-}
[2/3] incubator-fluo-recipes git commit: Updated ExportQ and CFM to
use new ObserverProvider API
Posted by kt...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloTranslatorTest.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloTranslatorTest.java b/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloTranslatorTest.java
new file mode 100644
index 0000000..d8a2836
--- /dev/null
+++ b/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloTranslatorTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.accumulo.export;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.recipes.accumulo.export.function.AccumuloTranslator;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AccumuloTranslatorTest {
+
+ public static Map<RowColumn, Bytes> genData(String key, Optional<String> val) {
+ if (!val.isPresent()) {
+ return Collections.emptyMap();
+ }
+ Map<RowColumn, Bytes> rcMap = new HashMap<>();
+ String data = val.get();
+ for (int i = 0; i < data.length(); i++) {
+ char c = data.charAt(i);
+ rcMap.put(new RowColumn("r:" + key, new Column("cf:" + c)), Bytes.of("v:" + c));
+ }
+ return rcMap;
+ }
+
+ public static void genMutations(String key, long seq, Optional<String> oldVal,
+ Optional<String> newVal, Consumer<Mutation> consumer) {
+ AccumuloTranslator.generateMutations(seq, genData(key, oldVal), genData(key, newVal), consumer);
+ }
+
+ public static Mutation makePut(String key, String val, long seq) {
+ Mutation m = new Mutation("r:" + key);
+ addPut(m, key, val, seq);
+ return m;
+ }
+
+ public static void addPut(Mutation m, String key, String val, long seq) {
+ m.put("cf:" + val, "", seq, "v:" + val);
+ }
+
+ public static Mutation makeDel(String key, String val, long seq) {
+ Mutation m = new Mutation("r:" + key);
+ addDel(m, key, val, seq);
+ return m;
+ }
+
+ public static void addDel(Mutation m, String key, String val, long seq) {
+ m.putDelete("cf:" + val, "", seq);
+ }
+
+ @Test
+ public void testDifferenceExport() {
+ final Collection<Mutation> mutations = new ArrayList<>();
+ Consumer<Mutation> consumer = m -> mutations.add(m);
+
+ genMutations("k1", 1, Optional.empty(), Optional.of("a"), consumer);
+ Assert.assertEquals(1, mutations.size());
+ Assert.assertTrue(mutations.contains(makePut("k1", "a", 1)));
+ mutations.clear();
+
+ genMutations("k2", 2, Optional.of("ab"), Optional.of("ab"), consumer);
+ Assert.assertEquals(0, mutations.size());
+ mutations.clear();
+
+ genMutations("k2", 2, Optional.of("b"), Optional.of("ab"), consumer);
+ Assert.assertEquals(1, mutations.size());
+ Assert.assertTrue(mutations.contains(makePut("k2", "a", 2)));
+ mutations.clear();
+
+ genMutations("k3", 3, Optional.of("c"), Optional.of("d"), consumer);
+ Assert.assertEquals(1, mutations.size());
+ Mutation m = makeDel("k3", "c", 3);
+ addPut(m, "k3", "d", 3);
+ Assert.assertTrue(mutations.contains(m));
+ mutations.clear();
+
+ genMutations("k4", 4, Optional.of("e"), Optional.empty(), consumer);
+ Assert.assertEquals(1, mutations.size());
+ Assert.assertTrue(mutations.contains(makeDel("k4", "e", 4)));
+ mutations.clear();
+
+ genMutations("k5", 5, Optional.of("ef"), Optional.of("fg"), consumer);
+ Assert.assertEquals(1, mutations.size());
+ m = makeDel("k5", "e", 5);
+ addPut(m, "k5", "g", 5);
+ Assert.assertTrue(mutations.contains(m));
+ mutations.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeImpl.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeImpl.java
new file mode 100644
index 0000000..c9fd9e0
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeImpl.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Optional;
+
+// intentionally package private
+class ChangeImpl<K, V> implements ChangeObserver.Change<K, V> {
+
+ private final K key;
+ private final Optional<V> oldValue;
+ private final Optional<V> newValue;
+
+ ChangeImpl(K key, Optional<V> oldValue, Optional<V> newValue) {
+ this.key = key;
+ this.oldValue = oldValue;
+ this.newValue = newValue;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public Optional<V> getNewValue() {
+ return newValue;
+ }
+
+ @Override
+ public Optional<V> getOldValue() {
+ return oldValue;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeObserver.java
new file mode 100644
index 0000000..4e39081
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/ChangeObserver.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Optional;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.recipes.core.export.ExportQueue;
+
+/**
+ * {@link CombineQueue} uses this interface to notify of changes to a keys value. It provides the
+ * new and old value for a key. For efficiency, the {@link CombineQueue} processes batches of key
+ * updates at once. It is strongly advised to only use the passed in transaction for writes that are
+ * unlikely to collide. If one write collides, then it will cause the whole batch to fail. Examples
+ * of writes that will not collide are updating an {@link ExportQueue} or another
+ * {@link CombineQueue}.
+ *
+ * <p>
+ * It was advised to only do writes because reads for each key will slow down processing a batch. If
+ * reading data is necessary then consider doing batch reads.
+ *
+ * @since 1.1.0
+ */
+@FunctionalInterface
+public interface ChangeObserver<K, V> {
+
+ /**
+ * @since 1.1.0
+ */
+ public static interface Change<K2, V2> {
+ public K2 getKey();
+
+ public Optional<V2> getNewValue();
+
+ public Optional<V2> getOldValue();
+ }
+
+ void process(TransactionBase tx, Iterable<Change<K, V>> changes);
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueue.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueue.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueue.java
new file mode 100644
index 0000000..265a0f9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueue.java
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.io.Serializable;
+import java.util.Map;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.observer.ObserverProvider.Registry;
+import org.apache.fluo.recipes.core.common.TableOptimizations;
+import org.apache.fluo.recipes.core.common.TableOptimizations.TableOptimizationsFactory;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+/**
+ * See the project level documentation for information about this recipe.
+ *
+ * @since 1.1.0
+ */
+public interface CombineQueue<K, V> {
+
+ /**
+ * Queues updates for this combine queue. These updates will be made by an Observer executing in
+ * another transaction. This method will not collide with other transaction queuing updates for
+ * the same keys.
+ *
+ * @param tx This transaction will be used to make the updates.
+ * @param updates The keys in the map should correspond to keys in the collision free map being
+ * updated. The values in the map will be queued for updating.
+ */
+ public void addAll(TransactionBase tx, Map<K, V> updates);
+
+ /**
+ * Used to register a Fluo Observer that processes updates to this combine queue. If this is not
+ * called, then updates will never be processed.
+ *
+ */
+ public void registerObserver(Registry obsRegistry, Combiner<K, V> combiner,
+ ChangeObserver<K, V> updateObserver);
+
+ /**
+ * Get a combiner queue instance.
+ *
+ * @param combineQueueId This should be the same id passed to {@link #configure(String)} before
+ * initializing Fluo.
+ * @param appConfig Application configuration obtained from
+ * {@code FluoClient.getAppConfiguration()},
+ * {@code FluoConfiguration.getAppConfiguration()}, or
+ * {@code ObserverProvider.Context.getAppConfiguration()}
+ */
+ public static <K2, V2> CombineQueue<K2, V2> getInstance(String combineQueueId,
+ SimpleConfiguration appConfig) {
+ try {
+ return new CombineQueueImpl<>(combineQueueId, appConfig);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Part of a fluent API for configuring a combine queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg1 {
+ public FluentArg2 keyType(String keyType);
+
+ public FluentArg2 keyType(Class<?> keyType);
+ }
+
+ /**
+ * Part of a fluent API for configuring a combine queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg2 {
+ public FluentArg3 valueType(String valType);
+
+ public FluentArg3 valueType(Class<?> valType);
+ }
+
+ /**
+ * Part of a fluent API for configuring a combine queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg3 {
+ FluentOptions buckets(int numBuckets);
+ }
+
+ /**
+ * Part of a fluent API for configuring a combine queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentOptions {
+ /**
+ * Sets a limit on the amount of serialized updates to read into memory. Additional memory will
+ * be used to actually deserialize and process the updates. This limit does not account for
+ * object overhead in java, which can be significant.
+ *
+ * <p>
+ * The way memory read is calculated is by summing the length of serialized key and value byte
+ * arrays. Once this sum exceeds the configured memory limit, no more update key values are
+ * processed in the current transaction. When not everything is processed, the observer
+ * processing updates will notify itself causing another transaction to continue processing
+ * later
+ */
+ public FluentOptions bufferSize(long bufferSize);
+
+ /**
+ * Sets the number of buckets per tablet to generate. This affects how many split points will be
+ * generated when optimizing the Accumulo table.
+ */
+ public FluentOptions bucketsPerTablet(int bucketsPerTablet);
+
+ /**
+ * Adds properties to the Fluo application configuration for this CombineQueue.
+ */
+ public void save(FluoConfiguration fluoConfig);
+ }
+
+ /**
+ * Call this method before initializing Fluo to configure a combine queue.
+ *
+ * @param combineQueueId An id that uniquely identifies a combine queue. This id is used in the
+ * keys in the Fluo table and in the keys in the Fluo application configuration.
+ * @return A Fluent configurator.
+ */
+ public static FluentArg1 configure(String combineQueueId) {
+ return new CqConfigurator(combineQueueId);
+ }
+
+ /**
+ * @since 1.1.0
+ */
+ public static interface Initializer<K2, V2> extends Serializable {
+ public RowColumnValue convert(K2 key, V2 val);
+ }
+
+ /**
+ * A {@link CombineQueue} stores data in its own data format in the Fluo table. When initializing
+ * a Fluo table with something like Map Reduce or Spark, data will need to be written in this
+ * format. Thats the purpose of this method, it provides a simple class that can do this
+ * conversion.
+ */
+ public static <K2, V2> Initializer<K2, V2> getInitializer(String cqId, int numBuckets,
+ SimpleSerializer serializer) {
+ return new InitializerImpl<>(cqId, numBuckets, serializer);
+ }
+
+ /**
+ * @since 1.1.0
+ */
+ public static class Optimizer implements TableOptimizationsFactory {
+ /**
+ * Return suggested Fluo table optimizations for the specified combine queue.
+ *
+ * @param appConfig Must pass in the application configuration obtained from
+ * {@code FluoClient.getAppConfiguration()} or
+ * {@code FluoConfiguration.getAppConfiguration()}
+ */
+ @Override
+ public TableOptimizations getTableOptimizations(String cqId, SimpleConfiguration appConfig) {
+ return CqOptimizer.getTableOptimizations(cqId, appConfig);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueueImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueueImpl.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueueImpl.java
new file mode 100644
index 0000000..bf97a5e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CombineQueueImpl.java
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.Sets;
+import com.google.common.hash.Hashing;
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Bytes.BytesBuilder;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.observer.Observer.NotificationType;
+import org.apache.fluo.api.observer.ObserverProvider.Registry;
+import org.apache.fluo.recipes.core.combine.ChangeObserver.Change;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+// intentionally package private
+class CombineQueueImpl<K, V> implements CombineQueue<K, V> {
+ static final Column DATA_COLUMN = new Column("data", "current");
+ static final Column UPDATE_COL = new Column("u", "v");
+ static final Column NEXT_COL = new Column("u", "next");
+
+ private Bytes updatePrefix;
+ private Bytes dataPrefix;
+ private Column notifyColumn;
+
+ private final Class<K> keyType;
+ private final Class<V> valType;
+ private final int numBuckets;
+ private final long bufferSize;
+ private SimpleSerializer serializer;
+
+ @SuppressWarnings("unchecked")
+ CombineQueueImpl(String cqId, SimpleConfiguration appConfig) throws Exception {
+ this.updatePrefix = Bytes.of(cqId + ":u:");
+ this.dataPrefix = Bytes.of(cqId + ":d:");
+ this.notifyColumn = new Column("fluoRecipes", "cfm:" + cqId);
+ this.keyType =
+ (Class<K>) getClass().getClassLoader()
+ .loadClass(CqConfigurator.getKeyType(cqId, appConfig));
+ this.valType =
+ (Class<V>) getClass().getClassLoader().loadClass(
+ CqConfigurator.getValueType(cqId, appConfig));
+ this.numBuckets = CqConfigurator.getNumBucket(cqId, appConfig);
+ this.bufferSize = CqConfigurator.getBufferSize(cqId, appConfig);
+ this.serializer = SimpleSerializer.getInstance(appConfig);
+ }
+
+ private static byte[] encSeq(long l) {
+ byte[] ret = new byte[8];
+ ret[0] = (byte) (l >>> 56);
+ ret[1] = (byte) (l >>> 48);
+ ret[2] = (byte) (l >>> 40);
+ ret[3] = (byte) (l >>> 32);
+ ret[4] = (byte) (l >>> 24);
+ ret[5] = (byte) (l >>> 16);
+ ret[6] = (byte) (l >>> 8);
+ ret[7] = (byte) (l >>> 0);
+ return ret;
+ }
+
+ static String genBucketId(int bucket, int maxBucket) {
+ Preconditions.checkArgument(bucket >= 0);
+ Preconditions.checkArgument(maxBucket > 0);
+
+ int bits = 32 - Integer.numberOfLeadingZeros(maxBucket);
+ int bucketLen = bits / 4 + (bits % 4 > 0 ? 1 : 0);
+
+ return Strings.padStart(Integer.toHexString(bucket), bucketLen, '0');
+ }
+
+ @Override
+ public void addAll(TransactionBase tx, Map<K, V> updates) {
+ Preconditions.checkState(numBuckets > 0, "Not initialized");
+
+ Set<String> buckets = new HashSet<>();
+
+ BytesBuilder rowBuilder = Bytes.builder();
+ rowBuilder.append(updatePrefix);
+ int prefixLength = rowBuilder.getLength();
+
+ byte[] startTs = encSeq(tx.getStartTimestamp());
+
+ for (Entry<K, V> entry : updates.entrySet()) {
+ byte[] k = serializer.serialize(entry.getKey());
+ int hash = Hashing.murmur3_32().hashBytes(k).asInt();
+ String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
+
+ // reset to the common row prefix
+ rowBuilder.setLength(prefixLength);
+
+ Bytes row = rowBuilder.append(bucketId).append(':').append(k).append(startTs).toBytes();
+ Bytes val = Bytes.of(serializer.serialize(entry.getValue()));
+
+ // TODO set if not exists would be comforting here.... but
+ // collisions on bucketId+key+uuid should never occur
+ tx.set(row, UPDATE_COL, val);
+
+ buckets.add(bucketId);
+ }
+
+ for (String bucketId : buckets) {
+ rowBuilder.setLength(prefixLength);
+ rowBuilder.append(bucketId).append(':');
+
+ Bytes row = rowBuilder.toBytes();
+
+ tx.setWeakNotification(row, notifyColumn);
+ }
+ }
+
+ private Map<Bytes, Map<Column, Bytes>> getCurrentValues(TransactionBase tx, BytesBuilder prefix,
+ Set<Bytes> keySet) {
+
+ Set<Bytes> rows = new HashSet<>();
+
+ int prefixLen = prefix.getLength();
+ for (Bytes key : keySet) {
+ prefix.setLength(prefixLen);
+ rows.add(prefix.append(key).toBytes());
+ }
+
+ try {
+ return tx.get(rows, Collections.singleton(DATA_COLUMN));
+ } catch (IllegalArgumentException e) {
+ System.out.println(rows.size());
+ throw e;
+ }
+ }
+
+ private V deserVal(Bytes val) {
+ return serializer.deserialize(val.toArray(), valType);
+ }
+
+ private Bytes getKeyFromUpdateRow(Bytes prefix, Bytes row) {
+ return row.subSequence(prefix.length(), row.length() - 8);
+ }
+
+ void process(TransactionBase tx, Bytes ntfyRow, Column col, Combiner<K, V> combiner,
+ ChangeObserver<K, V> changeObserver) throws Exception {
+
+ Preconditions.checkState(ntfyRow.startsWith(updatePrefix));
+
+ Bytes nextKey = tx.get(ntfyRow, NEXT_COL);
+
+ Span span;
+
+ if (nextKey != null) {
+ Bytes startRow =
+ Bytes.builder(ntfyRow.length() + nextKey.length()).append(ntfyRow).append(nextKey)
+ .toBytes();
+ Span tmpSpan = Span.prefix(ntfyRow);
+ Span nextSpan =
+ new Span(new RowColumn(startRow, UPDATE_COL), false, tmpSpan.getEnd(),
+ tmpSpan.isEndInclusive());
+ span = nextSpan;
+ } else {
+ span = Span.prefix(ntfyRow);
+ }
+
+ Iterator<RowColumnValue> iter = tx.scanner().over(span).fetch(UPDATE_COL).build().iterator();
+
+ Map<Bytes, List<Bytes>> updates = new HashMap<>();
+
+ long approxMemUsed = 0;
+
+ Bytes partiallyReadKey = null;
+ boolean setNextKey = false;
+
+ if (iter.hasNext()) {
+ Bytes lastKey = null;
+ while (iter.hasNext() && approxMemUsed < bufferSize) {
+ RowColumnValue rcv = iter.next();
+ Bytes curRow = rcv.getRow();
+
+ tx.delete(curRow, UPDATE_COL);
+
+ Bytes serializedKey = getKeyFromUpdateRow(ntfyRow, curRow);
+ lastKey = serializedKey;
+
+ List<Bytes> updateList = updates.get(serializedKey);
+ if (updateList == null) {
+ updateList = new ArrayList<>();
+ updates.put(serializedKey, updateList);
+ }
+
+ Bytes val = rcv.getValue();
+ updateList.add(val);
+
+ approxMemUsed += curRow.length();
+ approxMemUsed += val.length();
+ }
+
+ if (iter.hasNext()) {
+ RowColumnValue rcv = iter.next();
+ Bytes curRow = rcv.getRow();
+
+ // check if more updates for last key
+ if (getKeyFromUpdateRow(ntfyRow, curRow).equals(lastKey)) {
+ // there are still more updates for this key
+ partiallyReadKey = lastKey;
+
+ // start next time at the current key
+ tx.set(ntfyRow, NEXT_COL, partiallyReadKey);
+ } else {
+ // start next time at the next possible key
+ Bytes nextPossible =
+ Bytes.builder(lastKey.length() + 1).append(lastKey).append(0).toBytes();
+ tx.set(ntfyRow, NEXT_COL, nextPossible);
+ }
+
+ setNextKey = true;
+ } else if (nextKey != null) {
+ // clear nextKey
+ tx.delete(ntfyRow, NEXT_COL);
+ }
+ } else if (nextKey != null) {
+ tx.delete(ntfyRow, NEXT_COL);
+ }
+
+ if (nextKey != null || setNextKey) {
+ // If not all data was read need to run again in the future. If scanning was started in the
+ // middle of the bucket, its possible there is new data before nextKey that still needs to be
+ // processed. If scanning stopped before reading the entire bucket there may be data after the
+ // stop point.
+ tx.setWeakNotification(ntfyRow, col);
+ }
+
+
+
+ BytesBuilder rowBuilder = Bytes.builder();
+ rowBuilder.append(dataPrefix);
+ rowBuilder.append(ntfyRow.subSequence(updatePrefix.length(), ntfyRow.length()));
+ int rowPrefixLen = rowBuilder.getLength();
+
+ Set<Bytes> keysToFetch = updates.keySet();
+ if (partiallyReadKey != null) {
+ final Bytes prk = partiallyReadKey;
+ keysToFetch = Sets.filter(keysToFetch, b -> !b.equals(prk));
+ }
+ Map<Bytes, Map<Column, Bytes>> currentVals = getCurrentValues(tx, rowBuilder, keysToFetch);
+
+ ArrayList<Change<K, V>> updatesToReport = new ArrayList<>(updates.size());
+
+ for (Entry<Bytes, List<Bytes>> entry : updates.entrySet()) {
+ rowBuilder.setLength(rowPrefixLen);
+ Bytes currentValueRow = rowBuilder.append(entry.getKey()).toBytes();
+ Bytes currVal =
+ currentVals.getOrDefault(currentValueRow, Collections.emptyMap()).get(DATA_COLUMN);
+
+ K kd = serializer.deserialize(entry.getKey().toArray(), keyType);
+
+ if (partiallyReadKey != null && partiallyReadKey.equals(entry.getKey())) {
+ // not all updates were read for this key, so requeue the combined updates as an update
+ Optional<V> nv = combiner.combine(new InputImpl<>(kd, this::deserVal, entry.getValue()));
+ if (nv.isPresent()) {
+ addAll(tx, Collections.singletonMap(kd, nv.get()));
+ }
+ } else {
+ Optional<V> nv =
+ combiner.combine(new InputImpl<>(kd, this::deserVal, currVal, entry.getValue()));
+ Bytes newVal = nv.isPresent() ? Bytes.of(serializer.serialize(nv.get())) : null;
+ if (newVal != null ^ currVal != null || (currVal != null && !currVal.equals(newVal))) {
+ if (newVal == null) {
+ tx.delete(currentValueRow, DATA_COLUMN);
+ } else {
+ tx.set(currentValueRow, DATA_COLUMN, newVal);
+ }
+
+ Optional<V> cvd = Optional.ofNullable(currVal).map(this::deserVal);
+ updatesToReport.add(new ChangeImpl<>(kd, cvd, nv));
+ }
+ }
+ }
+
+ // TODO could clear these as converted to objects to avoid double memory usage
+ updates.clear();
+ currentVals.clear();
+
+ if (updatesToReport.size() > 0) {
+ changeObserver.process(tx, updatesToReport);
+ }
+ }
+
+ @Override
+ public void registerObserver(Registry obsRegistry, Combiner<K, V> combiner,
+ ChangeObserver<K, V> changeObserver) {
+ obsRegistry.register(notifyColumn, NotificationType.WEAK,
+ (tx, row, col) -> process(tx, row, col, combiner, changeObserver));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/Combiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/Combiner.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/Combiner.java
new file mode 100644
index 0000000..233ec84
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/Combiner.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Iterator;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+/**
+ * This class was created as an alternative to {@link Combiner}. It supports easy and efficient use
+ * of java streams when implementing combiners using lambdas.
+ *
+ * @since 1.1.0
+ */
+@FunctionalInterface
+public interface Combiner<K, V> {
+
+ /**
+ *
+ * @since 1.1.0
+ */
+ public static interface Input<KI, VI> extends Iterable<VI> {
+ KI getKey();
+
+ Stream<VI> stream();
+
+ Iterator<VI> iterator();
+ }
+
+ /**
+ * This function is called to combine the current value of a key with updates that were queued for
+ * the key. See the collision free map project level documentation for more information.
+ *
+ * @return Then new value for the key. Returning Optional.empty() will cause the key to be
+ * deleted.
+ */
+ Optional<V> combine(Input<K, V> input);
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqConfigurator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqConfigurator.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqConfigurator.java
new file mode 100644
index 0000000..40c5b37
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqConfigurator.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Objects;
+
+import com.google.common.base.Preconditions;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.recipes.core.combine.CombineQueue.FluentArg1;
+import org.apache.fluo.recipes.core.combine.CombineQueue.FluentArg2;
+import org.apache.fluo.recipes.core.combine.CombineQueue.FluentArg3;
+import org.apache.fluo.recipes.core.combine.CombineQueue.FluentOptions;
+import org.apache.fluo.recipes.core.common.RowRange;
+import org.apache.fluo.recipes.core.common.TableOptimizations;
+import org.apache.fluo.recipes.core.common.TransientRegistry;
+
+// this class intentionally package private
+class CqConfigurator implements FluentArg1, FluentArg2, FluentArg3, FluentOptions {
+
+ static final String UPDATE_RANGE_END = ":u:~";
+
+ static final String DATA_RANGE_END = ":d:~";
+
+ int numBuckets;
+ Integer bucketsPerTablet = null;
+
+ Long bufferSize;
+
+ String keyType;
+ String valueType;
+ String cqId;
+
+ static final int DEFAULT_BUCKETS_PER_TABLET = 10;
+
+ static final long DEFAULT_BUFFER_SIZE = 1 << 22;
+
+ static final String PREFIX = "recipes.cfm.";
+
+ CqConfigurator(String id) {
+ Objects.requireNonNull(id);
+ Preconditions.checkArgument(!id.contains(":"), "Combine queue id cannot contain ':'");
+ this.cqId = id;
+ }
+
+ @Override
+ public FluentOptions buckets(int numBuckets) {
+ Preconditions.checkArgument(numBuckets > 0);
+ this.numBuckets = numBuckets;
+ return this;
+ }
+
+ @Override
+ public FluentArg3 valueType(String valType) {
+ this.valueType = Objects.requireNonNull(valType);
+ return this;
+ }
+
+
+ @Override
+ public FluentArg3 valueType(Class<?> valType) {
+ this.valueType = valType.getName();
+ return this;
+ }
+
+ @Override
+ public FluentArg2 keyType(String keyType) {
+ this.keyType = Objects.requireNonNull(keyType);
+ return this;
+ }
+
+ @Override
+ public FluentArg2 keyType(Class<?> keyType) {
+ this.keyType = keyType.getName();
+ return this;
+ }
+
+ @Override
+ public FluentOptions bufferSize(long bufferSize) {
+ Preconditions.checkArgument(bufferSize > 0, "Buffer size must be positive");
+ this.bufferSize = bufferSize;
+ return this;
+ }
+
+ @Override
+ public FluentOptions bucketsPerTablet(int bucketsPerTablet) {
+ Preconditions.checkArgument(bucketsPerTablet > 0, "bucketsPerTablet is <= 0 : "
+ + bucketsPerTablet);
+ this.bucketsPerTablet = bucketsPerTablet;
+ return this;
+ }
+
+ @Override
+ public void save(FluoConfiguration fluoConfig) {
+ SimpleConfiguration appConfig = fluoConfig.getAppConfiguration();
+ appConfig.setProperty(PREFIX + cqId + ".buckets", numBuckets + "");
+ appConfig.setProperty(PREFIX + cqId + ".key", keyType);
+ appConfig.setProperty(PREFIX + cqId + ".val", valueType);
+ if (bufferSize != null) {
+ appConfig.setProperty(PREFIX + cqId + ".bufferSize", bufferSize);
+ }
+ if (bucketsPerTablet != null) {
+ appConfig.setProperty(PREFIX + cqId + ".bucketsPerTablet", bucketsPerTablet);
+ }
+
+ Bytes dataRangeEnd = Bytes.of(cqId + DATA_RANGE_END);
+ Bytes updateRangeEnd = Bytes.of(cqId + UPDATE_RANGE_END);
+
+ new TransientRegistry(fluoConfig.getAppConfiguration()).addTransientRange("cfm." + cqId,
+ new RowRange(dataRangeEnd, updateRangeEnd));
+
+ TableOptimizations.registerOptimization(appConfig, cqId, CombineQueue.Optimizer.class);
+ }
+
+ static long getBufferSize(String cqId, SimpleConfiguration appConfig) {
+ return appConfig.getLong(PREFIX + cqId + ".bufferSize", DEFAULT_BUFFER_SIZE);
+ }
+
+ static String getValueType(String cqId, SimpleConfiguration appConfig) {
+ return appConfig.getString(PREFIX + cqId + ".val");
+ }
+
+ static String getKeyType(String cqId, SimpleConfiguration appConfig) {
+ return appConfig.getString(PREFIX + cqId + ".key");
+ }
+
+ static int getBucketsPerTablet(String cqId, SimpleConfiguration appConfig) {
+ return appConfig.getInt(PREFIX + cqId + ".bucketsPerTablet", DEFAULT_BUCKETS_PER_TABLET);
+ }
+
+ static int getNumBucket(String cqId, SimpleConfiguration appConfig) {
+ return appConfig.getInt(PREFIX + cqId + ".buckets");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqOptimizer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqOptimizer.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqOptimizer.java
new file mode 100644
index 0000000..b9e02d6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/CqOptimizer.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Bytes.BytesBuilder;
+import org.apache.fluo.recipes.core.common.TableOptimizations;
+
+// This class intentionally package private.
+class CqOptimizer {
+
+ public static TableOptimizations getTableOptimizations(String cqId, SimpleConfiguration appConfig) {
+ int numBuckets = CqConfigurator.getNumBucket(cqId, appConfig);
+ int bpt = CqConfigurator.getBucketsPerTablet(cqId, appConfig);
+
+ BytesBuilder rowBuilder = Bytes.builder();
+ rowBuilder.append(cqId);
+
+ List<Bytes> dataSplits = new ArrayList<>();
+ for (int i = bpt; i < numBuckets; i += bpt) {
+ String bucketId = CombineQueueImpl.genBucketId(i, numBuckets);
+ rowBuilder.setLength(cqId.length());
+ dataSplits.add(rowBuilder.append(":d:").append(bucketId).toBytes());
+ }
+ Collections.sort(dataSplits);
+
+ List<Bytes> updateSplits = new ArrayList<>();
+ for (int i = bpt; i < numBuckets; i += bpt) {
+ String bucketId = CombineQueueImpl.genBucketId(i, numBuckets);
+ rowBuilder.setLength(cqId.length());
+ updateSplits.add(rowBuilder.append(":u:").append(bucketId).toBytes());
+ }
+ Collections.sort(updateSplits);
+
+ Bytes dataRangeEnd = Bytes.of(cqId + CqConfigurator.DATA_RANGE_END);
+ Bytes updateRangeEnd = Bytes.of(cqId + CqConfigurator.UPDATE_RANGE_END);
+
+ List<Bytes> splits = new ArrayList<>();
+ splits.add(dataRangeEnd);
+ splits.add(updateRangeEnd);
+ splits.addAll(dataSplits);
+ splits.addAll(updateSplits);
+
+ TableOptimizations tableOptim = new TableOptimizations();
+ tableOptim.setSplits(splits);
+
+ tableOptim.setTabletGroupingRegex(Pattern.quote(cqId + ":") + "[du]:");
+
+ return tableOptim;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InitializerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InitializerImpl.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InitializerImpl.java
new file mode 100644
index 0000000..5573ed8
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InitializerImpl.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import com.google.common.hash.Hashing;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Bytes.BytesBuilder;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.recipes.core.combine.CombineQueue.Initializer;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+// intentionally package private
+class InitializerImpl<K, V> implements Initializer<K, V> {
+ private static final long serialVersionUID = 1L;
+
+ private Bytes dataPrefix;
+
+ private SimpleSerializer serializer;
+
+ private int numBuckets = -1;
+
+ InitializerImpl(String cqId, int numBuckets, SimpleSerializer serializer) {
+ this.dataPrefix = Bytes.of(cqId + ":d:");
+ this.numBuckets = numBuckets;
+ this.serializer = serializer;
+ }
+
+ public RowColumnValue convert(K key, V val) {
+ byte[] k = serializer.serialize(key);
+ int hash = Hashing.murmur3_32().hashBytes(k).asInt();
+ String bucketId = CombineQueueImpl.genBucketId(Math.abs(hash % numBuckets), numBuckets);
+
+ BytesBuilder bb = Bytes.builder(dataPrefix.length() + bucketId.length() + 1 + k.length);
+ Bytes row = bb.append(dataPrefix).append(bucketId).append(':').append(k).toBytes();
+ byte[] v = serializer.serialize(val);
+
+ return new RowColumnValue(row, CombineQueueImpl.DATA_COLUMN, Bytes.of(v));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InputImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InputImpl.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InputImpl.java
new file mode 100644
index 0000000..aaca954
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/InputImpl.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import com.google.common.collect.Iterators;
+import org.apache.fluo.api.data.Bytes;
+
+// intentionally package priave
+class InputImpl<K, V> implements Combiner.Input<K, V> {
+ private K key;
+ private Collection<Bytes> valuesCollection;
+ private Bytes currentValue;
+ private Function<Bytes, V> valDeser;
+
+ InputImpl(K k, Function<Bytes, V> valDeser, Collection<Bytes> serializedValues) {
+ this.key = k;
+ this.valDeser = valDeser;
+ this.valuesCollection = serializedValues;
+ }
+
+ InputImpl(K k, Function<Bytes, V> valDeser, Bytes currentValue, Collection<Bytes> serializedValues) {
+ this(k, valDeser, serializedValues);
+ this.currentValue = currentValue;
+ }
+
+ @Override
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public Stream<V> stream() {
+ Stream<Bytes> bytesStream = valuesCollection.stream();
+ if (currentValue != null) {
+ bytesStream = Stream.concat(Stream.of(currentValue), bytesStream);
+ }
+
+ return bytesStream.map(valDeser);
+ }
+
+ @Override
+ public Iterator<V> iterator() {
+ Iterator<Bytes> bytesIter = valuesCollection.iterator();
+ if (currentValue != null) {
+ bytesIter = Iterators.concat(bytesIter, Iterators.singletonIterator(currentValue));
+ }
+ return Iterators.transform(bytesIter, valDeser::apply);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/SummingCombiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/SummingCombiner.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/SummingCombiner.java
new file mode 100644
index 0000000..2fd7044
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/combine/SummingCombiner.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.combine;
+
+import java.util.Optional;
+
+/**
+ * Sums long values and returns Optional.empty() when the sum is zero.
+ *
+ * @since 1.1.0
+ */
+public class SummingCombiner<K> implements Combiner<K, Long> {
+
+ @Override
+ public Optional<Long> combine(Input<K, Long> input) {
+ long sum = 0;
+ for (Long l : input) {
+ sum += l;
+ }
+
+ return sum == 0 ? Optional.empty() : Optional.of(sum);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
index 4882ee1..7292ff9 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
@@ -34,6 +34,7 @@ import org.apache.fluo.recipes.core.types.TypedTransactionBase;
/**
* This class encapsulates a buckets serialization code.
*/
+// This class intentionally package private.
class ExportBucket {
private static final String NOTIFICATION_CF = "fluoRecipes";
private static final String NOTIFICATION_CQ_PREFIX = "eq:";
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportEntry.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportEntry.java
index e680204..f5692aa 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportEntry.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportEntry.java
@@ -15,6 +15,7 @@
package org.apache.fluo.recipes.core.export;
+// This class intentionally package private.
class ExportEntry {
byte[] key;
long seq;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
index d086577..92756be 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
@@ -15,10 +15,6 @@
package org.apache.fluo.recipes.core.export;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-import com.google.common.collect.Iterators;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
@@ -28,76 +24,34 @@ import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
/**
* @since 1.0.0
+ * @deprecated since 1.1.0
*/
+@Deprecated
public class ExportObserver<K, V> extends AbstractObserver {
- private static class MemLimitIterator implements Iterator<ExportEntry> {
-
- private long memConsumed = 0;
- private long memLimit;
- private int extraPerKey;
- private Iterator<ExportEntry> source;
-
- public MemLimitIterator(Iterator<ExportEntry> input, long limit, int extraPerKey) {
- this.source = input;
- this.memLimit = limit;
- this.extraPerKey = extraPerKey;
- }
-
- @Override
- public boolean hasNext() {
- return memConsumed < memLimit && source.hasNext();
- }
-
- @Override
- public ExportEntry next() {
- if (!hasNext()) {
- throw new NoSuchElementException();
- }
- ExportEntry ee = source.next();
- memConsumed += ee.key.length + extraPerKey + ee.value.length;
- return ee;
- }
-
- @Override
- public void remove() {
- source.remove();
- }
- }
+ private ExportObserverImpl<K, V> eoi;
private String queueId;
- private Class<K> keyType;
- private Class<V> valType;
- SimpleSerializer serializer;
- private Exporter<K, V> exporter;
-
- private long memLimit;
protected String getQueueId() {
return queueId;
}
- SimpleSerializer getSerializer() {
- return serializer;
- }
-
@SuppressWarnings("unchecked")
@Override
public void init(Context context) throws Exception {
+
queueId = context.getObserverConfiguration().getString("queueId");
ExportQueue.Options opts = new ExportQueue.Options(queueId, context.getAppConfiguration());
// TODO defer loading classes... so that not done during fluo init
// TODO move class loading to centralized place... also attempt to check type params
- keyType = (Class<K>) getClass().getClassLoader().loadClass(opts.keyType);
- valType = (Class<V>) getClass().getClassLoader().loadClass(opts.valueType);
- exporter =
- getClass().getClassLoader().loadClass(opts.exporterType).asSubclass(Exporter.class)
- .newInstance();
-
- serializer = SimpleSerializer.getInstance(context.getAppConfiguration());
+ @SuppressWarnings("rawtypes")
+ Exporter exporter =
+ getClass().getClassLoader().loadClass(opts.fluentCfg.exporterType)
+ .asSubclass(Exporter.class).newInstance();
- memLimit = opts.getBufferSize();
+ SimpleSerializer serializer = SimpleSerializer.getInstance(context.getAppConfiguration());
exporter.init(new Exporter.Context() {
@@ -116,6 +70,11 @@ public class ExportObserver<K, V> extends AbstractObserver {
return context;
}
});
+
+
+ this.eoi =
+ new ExportObserverImpl<K, V>(queueId, opts.fluentCfg, serializer, exporter::processExports);
+
}
@Override
@@ -124,40 +83,7 @@ public class ExportObserver<K, V> extends AbstractObserver {
}
@Override
- public void process(TransactionBase tx, Bytes row, Column column) throws Exception {
- ExportBucket bucket = new ExportBucket(tx, row);
-
- Bytes continueRow = bucket.getContinueRow();
-
- Iterator<ExportEntry> input = bucket.getExportIterator(continueRow);
- MemLimitIterator memLimitIter = new MemLimitIterator(input, memLimit, 8 + queueId.length());
-
- Iterator<SequencedExport<K, V>> exportIterator =
- Iterators.transform(
- memLimitIter,
- ee -> new SequencedExport<>(serializer.deserialize(ee.key, keyType), serializer
- .deserialize(ee.value, valType), ee.seq));
-
- exportIterator = Iterators.consumingIterator(exportIterator);
-
- exporter.processExports(exportIterator);
-
- if (input.hasNext() || continueRow != null) {
- // not everything was processed so notify self OR new data may have been inserted above the
- // continue row
- bucket.notifyExportObserver();
- }
-
- if (input.hasNext()) {
- if (!memLimitIter.hasNext()) {
- // stopped because of mem limit... set continue key
- bucket.setContinueRow(input.next());
- continueRow = null;
- }
- }
-
- if (continueRow != null) {
- bucket.clearContinueRow();
- }
+ public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
+ eoi.process(tx, row, col);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserverImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserverImpl.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserverImpl.java
new file mode 100644
index 0000000..c54ecc0
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserverImpl.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.export;
+
+import java.util.Iterator;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.observer.Observer;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+import com.google.common.collect.Iterators;
+
+// This class intentionally package private.
+class ExportObserverImpl<K, V> implements Observer {
+
+ private String queueId;
+ private Class<K> keyType;
+ private Class<V> valType;
+ SimpleSerializer serializer;
+ private org.apache.fluo.recipes.core.export.function.Exporter<K, V> exporter;
+ private long memLimit;
+
+ @SuppressWarnings("unchecked")
+ ExportObserverImpl(String queueId, FluentConfigurator opts, SimpleSerializer serializer,
+ org.apache.fluo.recipes.core.export.function.Exporter<K, V> exportConsumer) throws Exception {
+ this.queueId = queueId;
+
+ // TODO move class loading to centralized place... also attempt to check type params
+ keyType = (Class<K>) getClass().getClassLoader().loadClass(opts.keyType);
+ valType = (Class<V>) getClass().getClassLoader().loadClass(opts.valueType);
+ exporter = exportConsumer;
+
+ this.serializer = serializer;
+
+ memLimit = opts.getBufferSize();
+ }
+
+ @Override
+ public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
+ ExportBucket bucket = new ExportBucket(tx, row);
+
+ Bytes continueRow = bucket.getContinueRow();
+
+ Iterator<ExportEntry> input = bucket.getExportIterator(continueRow);
+ MemLimitIterator memLimitIter = new MemLimitIterator(input, memLimit, 8 + queueId.length());
+
+ Iterator<SequencedExport<K, V>> exportIterator =
+ Iterators.transform(
+ memLimitIter,
+ ee -> new SequencedExport<>(serializer.deserialize(ee.key, keyType), serializer
+ .deserialize(ee.value, valType), ee.seq));
+
+ exportIterator = Iterators.consumingIterator(exportIterator);
+
+ exporter.export(exportIterator);
+
+ if (input.hasNext() || continueRow != null) {
+ // not everything was processed so notify self OR new data may have been inserted above the
+ // continue row
+ bucket.notifyExportObserver();
+ }
+
+ if (input.hasNext()) {
+ if (!memLimitIter.hasNext()) {
+ // stopped because of mem limit... set continue key
+ bucket.setContinueRow(input.next());
+ continueRow = null;
+ }
+ }
+
+ if (continueRow != null) {
+ bucket.clearContinueRow();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
index e30f7ce..e9618be 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
@@ -31,10 +31,11 @@ import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.ObserverSpecification;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.observer.Observer;
+import org.apache.fluo.api.observer.Observer.NotificationType;
+import org.apache.fluo.api.observer.ObserverProvider;
import org.apache.fluo.recipes.core.common.TableOptimizations;
import org.apache.fluo.recipes.core.common.TableOptimizations.TableOptimizationsFactory;
-import org.apache.fluo.recipes.core.common.RowRange;
-import org.apache.fluo.recipes.core.common.TransientRegistry;
import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
/**
@@ -42,23 +43,25 @@ import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
*/
public class ExportQueue<K, V> {
- private static final String RANGE_BEGIN = "#";
- private static final String RANGE_END = ":~";
+ static final String RANGE_BEGIN = "#";
+ static final String RANGE_END = ":~";
private int numBuckets;
private SimpleSerializer serializer;
private String queueId;
+ private FluentConfigurator opts;
// usage hint : could be created once in an observers init method
// usage hint : maybe have a queue for each type of data being exported???
// maybe less queues are
// more efficient though because more batching at export time??
- ExportQueue(Options opts, SimpleSerializer serializer) throws Exception {
+ ExportQueue(FluentConfigurator opts, SimpleSerializer serializer) throws Exception {
// TODO sanity check key type based on type params
// TODO defer creating classes until needed.. so that its not done during Fluo init
this.queueId = opts.queueId;
- this.numBuckets = opts.numBuckets;
+ this.numBuckets = opts.buckets;
this.serializer = serializer;
+ this.opts = opts;
}
public void add(TransactionBase tx, K key, V value) {
@@ -87,9 +90,11 @@ public class ExportQueue<K, V> {
}
}
+ // TODO maybe add for stream and interable
+
public static <K2, V2> ExportQueue<K2, V2> getInstance(String exportQueueId,
SimpleConfiguration appConfig) {
- Options opts = new Options(exportQueueId, appConfig);
+ FluentConfigurator opts = FluentConfigurator.load(exportQueueId, appConfig);
try {
return new ExportQueue<>(opts, SimpleSerializer.getInstance(appConfig));
} catch (Exception e) {
@@ -99,26 +104,100 @@ public class ExportQueue<K, V> {
}
/**
+ * Part of a fluent API for configuring a export queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg1 {
+ public FluentArg2 keyType(String keyType);
+
+ public FluentArg2 keyType(Class<?> keyType);
+ }
+
+ /**
+ * Part of a fluent API for configuring a export queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg2 {
+ public FluentArg3 valueType(String keyType);
+
+ public FluentArg3 valueType(Class<?> keyType);
+ }
+
+ /**
+ * Part of a fluent API for configuring a export queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentArg3 {
+ FluentOptions buckets(int numBuckets);
+ }
+
+ /**
+ * Part of a fluent API for configuring a export queue.
+ *
+ * @since 1.1.0
+ */
+ public static interface FluentOptions {
+ /**
+ * Sets a limit on the amount of serialized updates to read into memory. Additional memory will
+ * be used to actually deserialize and process the updates. This limit does not account for
+ * object overhead in java, which can be significant.
+ *
+ * <p>
+ * The way memory read is calculated is by summing the length of serialized key and value byte
+ * arrays. Once this sum exceeds the configured memory limit, no more export key values are
+ * processed in the current transaction. When not everything is processed, the observer
+ * processing exports will notify itself causing another transaction to continue processing
+ * later.
+ */
+ public FluentOptions bufferSize(long bufferSize);
+
+ /**
+ * Sets the number of buckets per tablet to generate. This affects how many split points will be
+ * generated when optimizing the Accumulo table.
+ */
+ public FluentOptions bucketsPerTablet(int bucketsPerTablet);
+
+ /**
+ * Adds properties to the Fluo application configuration for this CombineQueue.
+ */
+ public void save(FluoConfiguration fluoConfig);
+ }
+
+ /**
+ * A Fluent API for configuring an Export Queue. Use this method in conjunction with
+ * {@link #registerObserver(ObserverProvider.Registry, org.apache.fluo.recipes.core.export.function.Exporter)}
+ *
+ * @param exportQueueId An id that uniquely identifies an export queue. This id is used in the
+ * keys in the Fluo table and in the keys in the Fluo application configuration.
+ * @since 1.1.0
+ */
+ public static FluentArg1 configure(String exportQueueId) {
+ return new FluentConfigurator(Objects.requireNonNull(exportQueueId));
+ }
+
+ /**
* Call this method before initializing Fluo.
*
* @param fluoConfig The configuration that will be used to initialize fluo.
+ * @deprecated since 1.1.0 use {@link #configure(String)} and
+ * {@link #registerObserver(ObserverProvider.Registry, org.apache.fluo.recipes.core.export.function.Exporter)}
+ * instead.
*/
+ @Deprecated
public static void configure(FluoConfiguration fluoConfig, Options opts) {
SimpleConfiguration appConfig = fluoConfig.getAppConfiguration();
opts.save(appConfig);
fluoConfig.addObserver(new ObserverSpecification(ExportObserver.class.getName(), Collections
- .singletonMap("queueId", opts.queueId)));
-
- Bytes exportRangeStart = Bytes.of(opts.queueId + RANGE_BEGIN);
- Bytes exportRangeStop = Bytes.of(opts.queueId + RANGE_END);
-
- new TransientRegistry(fluoConfig.getAppConfiguration()).addTransientRange("exportQueue."
- + opts.queueId, new RowRange(exportRangeStart, exportRangeStop));
-
- TableOptimizations.registerOptimization(appConfig, opts.queueId, Optimizer.class);
+ .singletonMap("queueId", opts.fluentCfg.queueId)));
}
+ /**
+ * @since 1.0.0
+ */
public static class Optimizer implements TableOptimizationsFactory {
/**
@@ -130,7 +209,7 @@ public class ExportQueue<K, V> {
*/
@Override
public TableOptimizations getTableOptimizations(String queueId, SimpleConfiguration appConfig) {
- Options opts = new Options(queueId, appConfig);
+ FluentConfigurator opts = FluentConfigurator.load(queueId, appConfig);
List<Bytes> splits = new ArrayList<>();
@@ -141,8 +220,8 @@ public class ExportQueue<K, V> {
splits.add(exportRangeStop);
List<Bytes> exportSplits = new ArrayList<>();
- for (int i = opts.getBucketsPerTablet(); i < opts.numBuckets; i += opts.getBucketsPerTablet()) {
- exportSplits.add(ExportBucket.generateBucketRow(opts.queueId, i, opts.numBuckets));
+ for (int i = opts.getBucketsPerTablet(); i < opts.buckets; i += opts.getBucketsPerTablet()) {
+ exportSplits.add(ExportBucket.generateBucketRow(opts.queueId, i, opts.buckets));
}
Collections.sort(exportSplits);
splits.addAll(exportSplits);
@@ -160,47 +239,47 @@ public class ExportQueue<K, V> {
}
/**
+ * Registers an observer that will export queued data. Use this method in conjunction with
+ * {@link ExportQueue#configure(String)}.
+ *
+ * @since 1.1.0
+ */
+ public void registerObserver(ObserverProvider.Registry obsRegistry,
+ org.apache.fluo.recipes.core.export.function.Exporter<K, V> exporter) {
+ Preconditions
+ .checkState(
+ opts.exporterType == null,
+ "Expected exporter type not be set, it was set to %s. Cannot not use the old and new way of configuring "
+ + "exporters at the same time.", opts.exporterType);
+ Observer obs;
+ try {
+ obs = new ExportObserverImpl<K, V>(queueId, opts, serializer, exporter);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ obsRegistry.register(ExportBucket.newNotificationColumn(queueId), NotificationType.WEAK, obs);
+ }
+
+ /**
* @since 1.0.0
+ * @deprecated since 1.1.0 use {@link ExportQueue#configure(String)}
*/
public static class Options {
- private static final String PREFIX = "recipes.exportQueue.";
- static final long DEFAULT_BUFFER_SIZE = 1 << 20;
- static final int DEFAULT_BUCKETS_PER_TABLET = 10;
-
- int numBuckets;
- Integer bucketsPerTablet = null;
- Long bufferSize;
+ private static final String PREFIX = FluentConfigurator.PREFIX;
- String keyType;
- String valueType;
- String exporterType;
- String queueId;
+ FluentConfigurator fluentCfg;
SimpleConfiguration exporterConfig;
Options(String queueId, SimpleConfiguration appConfig) {
- this.queueId = queueId;
-
- this.numBuckets = appConfig.getInt(PREFIX + queueId + ".buckets");
- this.exporterType = appConfig.getString(PREFIX + queueId + ".exporter");
- this.keyType = appConfig.getString(PREFIX + queueId + ".key");
- this.valueType = appConfig.getString(PREFIX + queueId + ".val");
- this.bufferSize = appConfig.getLong(PREFIX + queueId + ".bufferSize", DEFAULT_BUFFER_SIZE);
- this.bucketsPerTablet =
- appConfig.getInt(PREFIX + queueId + ".bucketsPerTablet", DEFAULT_BUCKETS_PER_TABLET);
-
+ fluentCfg = FluentConfigurator.load(queueId, appConfig);
this.exporterConfig = appConfig.subset(PREFIX + queueId + ".exporterCfg");
}
public Options(String queueId, String exporterType, String keyType, String valueType,
int buckets) {
- Preconditions.checkArgument(buckets > 0);
-
- this.queueId = queueId;
- this.numBuckets = buckets;
- this.exporterType = exporterType;
- this.keyType = keyType;
- this.valueType = valueType;
+ this(queueId, keyType, valueType, buckets);
+ fluentCfg.exporterType = Objects.requireNonNull(exporterType);
}
public <K, V> Options(String queueId, Class<? extends Exporter<K, V>> exporter,
@@ -208,6 +287,14 @@ public class ExportQueue<K, V> {
this(queueId, exporter.getName(), keyType.getName(), valueType.getName(), buckets);
}
+ // intentionally package private
+ Options(String queueId, String keyType, String valueType, int buckets) {
+ Preconditions.checkArgument(buckets > 0);
+ this.fluentCfg =
+ (FluentConfigurator) new FluentConfigurator(queueId).keyType(keyType)
+ .valueType(valueType).buckets(buckets);
+ }
+
/**
* Sets a limit on the amount of serialized updates to read into memory. Additional memory will
* be used to actually deserialize and process the updates. This limit does not account for
@@ -221,17 +308,12 @@ public class ExportQueue<K, V> {
* later.
*/
public Options setBufferSize(long bufferSize) {
- Preconditions.checkArgument(bufferSize > 0, "Buffer size must be positive");
- this.bufferSize = bufferSize;
+ fluentCfg.bufferSize(bufferSize);
return this;
}
long getBufferSize() {
- if (bufferSize == null) {
- return DEFAULT_BUFFER_SIZE;
- }
-
- return bufferSize;
+ return fluentCfg.getBufferSize();
}
/**
@@ -240,24 +322,14 @@ public class ExportQueue<K, V> {
*
*/
public Options setBucketsPerTablet(int bucketsPerTablet) {
- Preconditions.checkArgument(bucketsPerTablet > 0, "bucketsPerTablet is <= 0 : "
- + bucketsPerTablet);
- this.bucketsPerTablet = bucketsPerTablet;
+ fluentCfg.bucketsPerTablet(bucketsPerTablet);
return this;
}
int getBucketsPerTablet() {
- if (bucketsPerTablet == null) {
- return DEFAULT_BUCKETS_PER_TABLET;
- }
-
- return bucketsPerTablet;
+ return fluentCfg.getBucketsPerTablet();
}
- /**
- * Sets exporter specific configuration. This configuration will be made available to an
- * {@link Exporter} via {@link Exporter.Context#getExporterConfiguration()}.
- */
public Options setExporterConfiguration(SimpleConfiguration config) {
Objects.requireNonNull(config);
this.exporterConfig = config;
@@ -273,28 +345,17 @@ public class ExportQueue<K, V> {
}
public String getQueueId() {
- return queueId;
+ return fluentCfg.queueId;
}
void save(SimpleConfiguration appConfig) {
- appConfig.setProperty(PREFIX + queueId + ".buckets", numBuckets + "");
- appConfig.setProperty(PREFIX + queueId + ".exporter", exporterType + "");
- appConfig.setProperty(PREFIX + queueId + ".key", keyType);
- appConfig.setProperty(PREFIX + queueId + ".val", valueType);
-
- if (bufferSize != null) {
- appConfig.setProperty(PREFIX + queueId + ".bufferSize", bufferSize);
- }
-
- if (bucketsPerTablet != null) {
- appConfig.setProperty(PREFIX + queueId + ".bucketsPerTablet", bucketsPerTablet);
- }
+ fluentCfg.save(appConfig);
if (exporterConfig != null) {
Iterator<String> keys = exporterConfig.getKeys();
while (keys.hasNext()) {
String key = keys.next();
- appConfig.setProperty(PREFIX + queueId + ".exporterCfg." + key,
+ appConfig.setProperty(PREFIX + fluentCfg.queueId + ".exporterCfg." + key,
exporterConfig.getRawString(key));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
index 66aa7e8..91802d6 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
@@ -19,10 +19,15 @@ import java.util.Iterator;
import org.apache.fluo.api.config.SimpleConfiguration;
import org.apache.fluo.api.observer.Observer;
+import org.apache.fluo.api.observer.ObserverProvider.Registry;
/**
* @since 1.0.0
+ * @deprecated since 1.1.0 replaced by
+ * {@link ExportQueue#registerObserver(Registry, org.apache.fluo.recipes.core.export.function.Exporter)}
+ * and {@link org.apache.fluo.recipes.core.export.function.Exporter}
*/
+@Deprecated
public abstract class Exporter<K, V> {
public interface Context {
@@ -71,6 +76,4 @@ public abstract class Exporter<K, V> {
*
*/
protected abstract void processExports(Iterator<SequencedExport<K, V>> exports);
-
- // TODO add close
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/FluentConfigurator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/FluentConfigurator.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/FluentConfigurator.java
new file mode 100644
index 0000000..718ec66
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/FluentConfigurator.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.export;
+
+import java.util.Objects;
+
+import com.google.common.base.Preconditions;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.recipes.core.common.RowRange;
+import org.apache.fluo.recipes.core.common.TableOptimizations;
+import org.apache.fluo.recipes.core.common.TransientRegistry;
+import org.apache.fluo.recipes.core.export.ExportQueue.FluentArg1;
+import org.apache.fluo.recipes.core.export.ExportQueue.FluentArg2;
+import org.apache.fluo.recipes.core.export.ExportQueue.FluentArg3;
+import org.apache.fluo.recipes.core.export.ExportQueue.FluentOptions;
+import org.apache.fluo.recipes.core.export.ExportQueue.Optimizer;
+
+// This class intentionally package private.
+class FluentConfigurator implements FluentArg1, FluentArg2, FluentArg3, FluentOptions {
+
+ static final long DEFAULT_BUFFER_SIZE = 1 << 20;
+ static final int DEFAULT_BUCKETS_PER_TABLET = 10;
+ static final String PREFIX = "recipes.exportQueue.";
+
+ String queueId;
+ Long bufferSize = null;
+ int buckets;
+ String valueType;
+ Integer bucketsPerTablet = null;
+ String keyType;
+ String exporterType;
+
+ FluentConfigurator(String queueId) {
+ this.queueId = queueId;
+ }
+
+ @Override
+ public FluentOptions bufferSize(long bufferSize) {
+ Preconditions.checkArgument(bufferSize > 0, "Buffer size must be positive");
+ this.bufferSize = bufferSize;
+ return this;
+ }
+
+ @Override
+ public FluentOptions bucketsPerTablet(int bucketsPerTablet) {
+ Preconditions.checkArgument(bucketsPerTablet > 0, "bucketsPerTablet is <= 0 : "
+ + bucketsPerTablet);
+ this.bucketsPerTablet = bucketsPerTablet;
+ return this;
+ }
+
+ void save(SimpleConfiguration appConfig) {
+ appConfig.setProperty(PREFIX + queueId + ".buckets", buckets + "");
+ appConfig.setProperty(PREFIX + queueId + ".key", keyType);
+ appConfig.setProperty(PREFIX + queueId + ".val", valueType);
+
+ if (exporterType != null) {
+ appConfig.setProperty(PREFIX + queueId + ".exporter", exporterType);
+ }
+
+ if (bufferSize != null) {
+ appConfig.setProperty(PREFIX + queueId + ".bufferSize", bufferSize);
+ }
+
+ if (bucketsPerTablet != null) {
+ appConfig.setProperty(PREFIX + queueId + ".bucketsPerTablet", bucketsPerTablet);
+ }
+
+ Bytes exportRangeStart = Bytes.of(queueId + ExportQueue.RANGE_BEGIN);
+ Bytes exportRangeStop = Bytes.of(queueId + ExportQueue.RANGE_END);
+
+ new TransientRegistry(appConfig).addTransientRange("exportQueue." + queueId, new RowRange(
+ exportRangeStart, exportRangeStop));
+
+ TableOptimizations.registerOptimization(appConfig, queueId, Optimizer.class);
+ }
+
+ @Override
+ public void save(FluoConfiguration fluoConfig) {
+ save(fluoConfig.getAppConfiguration());
+ }
+
+ static FluentConfigurator load(String queueId, SimpleConfiguration appConfig) {
+ FluentConfigurator fc = new FluentConfigurator(queueId);
+ fc.buckets = appConfig.getInt(PREFIX + queueId + ".buckets");
+ fc.keyType = appConfig.getString(PREFIX + queueId + ".key");
+ fc.valueType = appConfig.getString(PREFIX + queueId + ".val");
+ fc.bufferSize = appConfig.getLong(PREFIX + queueId + ".bufferSize", DEFAULT_BUFFER_SIZE);
+ fc.bucketsPerTablet =
+ appConfig.getInt(PREFIX + queueId + ".bucketsPerTablet", DEFAULT_BUCKETS_PER_TABLET);
+ fc.exporterType = appConfig.getString(PREFIX + queueId + ".exporter", null);
+ return fc;
+ }
+
+ long getBufferSize() {
+ if (bufferSize == null) {
+ return DEFAULT_BUFFER_SIZE;
+ }
+
+ return bufferSize;
+ }
+
+ int getBucketsPerTablet() {
+ if (bucketsPerTablet == null) {
+ return DEFAULT_BUCKETS_PER_TABLET;
+ }
+
+ return bucketsPerTablet;
+ }
+
+ @Override
+ public FluentOptions buckets(int numBuckets) {
+ Preconditions.checkArgument(numBuckets > 0);
+ this.buckets = numBuckets;
+ return this;
+ }
+
+ @Override
+ public FluentArg3 valueType(String valueType) {
+ this.valueType = Objects.requireNonNull(valueType);
+ return this;
+ }
+
+ @Override
+ public FluentArg3 valueType(Class<?> valueType) {
+ this.valueType = valueType.getName();
+ return this;
+ }
+
+ @Override
+ public FluentArg2 keyType(String keyType) {
+ this.keyType = Objects.requireNonNull(keyType);
+ return this;
+ }
+
+ @Override
+ public FluentArg2 keyType(Class<?> keyType) {
+ this.keyType = keyType.getName();
+ return this;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/MemLimitIterator.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/MemLimitIterator.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/MemLimitIterator.java
new file mode 100644
index 0000000..fe94bdc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/MemLimitIterator.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.export;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+// This class intentionally package private.
+class MemLimitIterator implements Iterator<ExportEntry> {
+
+ private long memConsumed = 0;
+ private long memLimit;
+ private int extraPerKey;
+ private Iterator<ExportEntry> source;
+
+ public MemLimitIterator(Iterator<ExportEntry> input, long limit, int extraPerKey) {
+ this.source = input;
+ this.memLimit = limit;
+ this.extraPerKey = extraPerKey;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return memConsumed < memLimit && source.hasNext();
+ }
+
+ @Override
+ public ExportEntry next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ ExportEntry ee = source.next();
+ memConsumed += ee.key.length + extraPerKey + ee.value.length;
+ return ee;
+ }
+
+ @Override
+ public void remove() {
+ source.remove();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/70a4239c/modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java
new file mode 100644
index 0000000..e6c14d5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/function/Exporter.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.fluo.recipes.core.export.function;
+
+import java.util.Iterator;
+import java.util.Objects;
+
+import org.apache.fluo.recipes.core.export.SequencedExport;
+
+/**
+ * Must be able to handle same key being exported multiple times and keys being exported out of
+ * order. The sequence number is meant to help with this.
+ *
+ * <p>
+ * If multiple export entries with the same key are passed in, then the entries with the same key
+ * will be consecutive and in ascending sequence order.
+ *
+ * <p>
+ * If the call to process exports is unexpectedly terminated, it will be called again later with at
+ * least the same data. For example suppose an exporter was passed the following entries.
+ *
+ * <ul>
+ * <li>key=0 sequence=9 value=abc
+ * <li>key=1 sequence=13 value=d
+ * <li>key=1 sequence=17 value=e
+ * <li>key=1 sequence=23 value=f
+ * <li>key=2 sequence=19 value=x
+ * </ul>
+ *
+ * <p>
+ * Assume the exporter exports some of these and then fails before completing all of them. The next
+ * time its called it will be passed what it saw before, but it could also be passed more.
+ *
+ * <ul>
+ * <li>key=0 sequence=9 value=abc
+ * <li>key=1 sequence=13 value=d
+ * <li>key=1 sequence=17 value=e
+ * <li>key=1 sequence=23 value=f
+ * <li>key=1 sequence=29 value=g
+ * <li>key=2 sequence=19 value=x
+ * <li>key=2 sequence=77 value=y
+ * </ul>
+ *
+ * @since 1.1.0
+ */
+@FunctionalInterface
+public interface Exporter<K, V> {
+
+ /**
+ * Performs this export operation.
+ *
+ * @param t the input argument
+ */
+ void export(Iterator<SequencedExport<K, V>> exports);
+
+ /**
+ * Returns a composed {@code Exporter} that exports, in sequence, to this then to {@code after}.
+ * If performing either export throws an exception, it is relayed to the caller of the composed
+ * operation. If performing this export operation throws an exception, the {@code after} export
+ * will not be performed.
+ *
+ * @param after the export operation to perform after this operation
+ * @return a composed {@code Exporter} that performs in sequence this export operation followed by
+ * the {@code after} export operation
+ * @throws NullPointerException if {@code after} is null
+ */
+ default Exporter<K, V> andThen(Exporter<K, V> after) {
+ Objects.requireNonNull(after);
+ return (Iterator<SequencedExport<K, V>> i) -> {
+ export(i);
+ after.export(i);
+ };
+ }
+}