You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by xi...@apache.org on 2016/12/15 08:24:57 UTC
storm git commit: STORM-2204: Adding caching capabilities in
HBaseLookupBolt closes apache/storm#1783
Repository: storm
Updated Branches:
refs/heads/master c08d61c74 -> 610d58d82
STORM-2204: Adding caching capabilities in HBaseLookupBolt closes apache/storm#1783
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/610d58d8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/610d58d8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/610d58d8
Branch: refs/heads/master
Commit: 610d58d829032854515939c912da142cbe8e2ccb
Parents: c08d61c
Author: vesense <be...@163.com>
Authored: Thu Dec 15 16:11:29 2016 +0800
Committer: vesense <be...@163.com>
Committed: Thu Dec 15 16:11:29 2016 +0800
----------------------------------------------------------------------
external/storm-hbase/README.md | 8 ++
external/storm-hbase/pom.xml | 8 +-
.../storm/hbase/bolt/AbstractHBaseBolt.java | 3 +-
.../storm/hbase/bolt/HBaseLookupBolt.java | 115 ++++++++++++-------
4 files changed, 92 insertions(+), 42 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/610d58d8/external/storm-hbase/README.md
----------------------------------------------------------------------
diff --git a/external/storm-hbase/README.md b/external/storm-hbase/README.md
index fd4d0ad..9efb663 100644
--- a/external/storm-hbase/README.md
+++ b/external/storm-hbase/README.md
@@ -170,6 +170,14 @@ The `HBaseLookupBolt` will use the mapper to get rowKey to lookup for. It will u
figure out which columns to include in the result and it will leverage the `HBaseRowToStormValueMapper` to get the
values to be emitted by the bolt.
+In addition, the `HBaseLookupBolt` supports bolt-side HBase result caching using an in-memory LRU cache using Caffeine. To enable caching:
+
+`hbase.cache.enable` - to enable caching (default false)
+
+`hbase.cache.ttl.seconds` - set time to live for LRU cache in seconds (default 300)
+
+`hbase.cache.size` - set size of the cache (default 1000)
+
You can look at an example topology LookupWordCount.java under `src/test/java`.
## Example: Persistent Word Count
A runnable example can be found in the `src/test/java` directory.
http://git-wip-us.apache.org/repos/asf/storm/blob/610d58d8/external/storm-hbase/pom.xml
----------------------------------------------------------------------
diff --git a/external/storm-hbase/pom.xml b/external/storm-hbase/pom.xml
index 5d1e0cd..293961d 100644
--- a/external/storm-hbase/pom.xml
+++ b/external/storm-hbase/pom.xml
@@ -38,6 +38,7 @@
<properties>
<hbase.version>1.1.0</hbase.version>
<hdfs.version>${hadoop.version}</hdfs.version>
+ <caffeine.version>2.3.5</caffeine.version>
</properties>
<dependencies>
@@ -88,5 +89,10 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>com.github.ben-manes.caffeine</groupId>
+ <artifactId>caffeine</artifactId>
+ <version>${caffeine.version}</version>
+ </dependency>
</dependencies>
-</project>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/610d58d8/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
index 76a0f8a..86c9761 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/AbstractHBaseBolt.java
@@ -37,8 +37,7 @@ import java.util.Map;
public abstract class AbstractHBaseBolt extends BaseRichBolt {
private static final Logger LOG = LoggerFactory.getLogger(AbstractHBaseBolt.class);
- protected OutputCollector collector;
-
+ protected transient OutputCollector collector;
protected transient HBaseClient hBaseClient;
protected String tableName;
protected HBaseMapper mapper;
http://git-wip-us.apache.org/repos/asf/storm/blob/610d58d8/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
index 58ef674..60faf7c 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
@@ -22,14 +22,22 @@ import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.TupleUtils;
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
import com.google.common.collect.Lists;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
import org.apache.commons.lang.Validate;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.storm.hbase.bolt.mapper.HBaseMapper;
import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
+import org.apache.storm.task.OutputCollector;
+import org.apache.storm.task.TopologyContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,51 +48,80 @@ import org.slf4j.LoggerFactory;
*
*/
public class HBaseLookupBolt extends AbstractHBaseBolt {
- private static final Logger LOG = LoggerFactory.getLogger(HBaseLookupBolt.class);
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseLookupBolt.class);
+
+ private HBaseValueMapper rowToTupleMapper;
+ private HBaseProjectionCriteria projectionCriteria;
+ private transient LoadingCache<byte[], Result> cache;
+ private transient boolean cacheEnabled;
- private HBaseValueMapper rowToTupleMapper;
+ public HBaseLookupBolt(String tableName, HBaseMapper mapper, HBaseValueMapper rowToTupleMapper) {
+ super(tableName, mapper);
+ Validate.notNull(rowToTupleMapper, "rowToTupleMapper can not be null");
+ this.rowToTupleMapper = rowToTupleMapper;
+ }
- private HBaseProjectionCriteria projectionCriteria;
+ public HBaseLookupBolt withConfigKey(String configKey) {
+ this.configKey = configKey;
+ return this;
+ }
- public HBaseLookupBolt(String tableName, HBaseMapper mapper, HBaseValueMapper rowToTupleMapper){
- super(tableName, mapper);
- Validate.notNull(rowToTupleMapper, "rowToTupleMapper can not be null");
- this.rowToTupleMapper = rowToTupleMapper;
- }
+ public HBaseLookupBolt withProjectionCriteria(HBaseProjectionCriteria projectionCriteria) {
+ this.projectionCriteria = projectionCriteria;
+ return this;
+ }
- public HBaseLookupBolt withConfigKey(String configKey){
- this.configKey = configKey;
- return this;
- }
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ @Override
+ public void prepare(Map config, TopologyContext topologyContext, OutputCollector collector) {
+ super.prepare(config, topologyContext, collector);
+ cacheEnabled = Boolean.parseBoolean(config.getOrDefault("hbase.cache.enable", "false").toString());
+ int cacheTTL = Integer.parseInt(config.getOrDefault("hbase.cache.ttl.seconds", "300").toString());
+ int maxCacheSize = Integer.parseInt(config.getOrDefault("hbase.cache.size", "1000").toString());
+ if (cacheEnabled) {
+ cache = Caffeine.newBuilder().maximumSize(maxCacheSize).expireAfterWrite(cacheTTL, TimeUnit.SECONDS)
+ .build(new CacheLoader<byte[], Result>() {
- public HBaseLookupBolt withProjectionCriteria(HBaseProjectionCriteria projectionCriteria) {
- this.projectionCriteria = projectionCriteria;
- return this;
- }
+ @Override
+ public Result load(byte[] rowKey) throws Exception {
+ Get get = hBaseClient.constructGetRequests(rowKey, projectionCriteria);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cache miss for key:" + new String(rowKey));
+ }
+ return hBaseClient.batchGet(Lists.newArrayList(get))[0];
+ }
- @Override
- public void execute(Tuple tuple) {
- if (TupleUtils.isTick(tuple)) {
- collector.ack(tuple);
- return;
- }
- byte[] rowKey = this.mapper.rowKey(tuple);
- Get get = hBaseClient.constructGetRequests(rowKey, projectionCriteria);
+ });
+ }
+ }
- try {
- Result result = hBaseClient.batchGet(Lists.newArrayList(get))[0];
- for(Values values : rowToTupleMapper.toValues(tuple, result)) {
- this.collector.emit(tuple, values);
- }
- this.collector.ack(tuple);
- } catch (Exception e) {
- this.collector.reportError(e);
- this.collector.fail(tuple);
- }
- }
+ @Override
+ public void execute(Tuple tuple) {
+ if (TupleUtils.isTick(tuple)) {
+ collector.ack(tuple);
+ return;
+ }
+ byte[] rowKey = this.mapper.rowKey(tuple);
+ Result result = null;
+ try {
+ if (cacheEnabled) {
+ result = cache.get(rowKey);
+ } else {
+ Get get = hBaseClient.constructGetRequests(rowKey, projectionCriteria);
+ result = hBaseClient.batchGet(Lists.newArrayList(get))[0];
+ }
+ for (Values values : rowToTupleMapper.toValues(tuple, result)) {
+ this.collector.emit(tuple, values);
+ }
+ this.collector.ack(tuple);
+ } catch (Exception e) {
+ this.collector.reportError(e);
+ this.collector.fail(tuple);
+ }
+ }
- @Override
- public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
- rowToTupleMapper.declareOutputFields(outputFieldsDeclarer);
- }
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+ rowToTupleMapper.declareOutputFields(outputFieldsDeclarer);
+ }
}