You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by pt...@apache.org on 2014/10/02 01:07:39 UTC
[19/26] git commit: refactor lookup interface
refactor lookup interface
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/fafda269
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/fafda269
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/fafda269
Branch: refs/heads/master
Commit: fafda26918fb7284f8a7ea69e4168f79775f4e42
Parents: 08d21a2
Author: P. Taylor Goetz <pt...@gmail.com>
Authored: Thu Sep 11 13:46:41 2014 -0400
Committer: P. Taylor Goetz <pt...@gmail.com>
Committed: Thu Sep 11 13:46:41 2014 -0400
----------------------------------------------------------------------
README.md | 6 +-
.../storm/hbase/bolt/HBaseLookupBolt.java | 6 +-
.../bolt/mapper/HBaseRowToStormValueMapper.java | 41 ------------
.../hbase/bolt/mapper/HBaseValueMapper.java | 41 ++++++++++++
.../storm/hbase/trident/state/HbaseState.java | 6 +-
.../storm/hbase/topology/LookupWordCount.java | 2 +-
.../WordCountRowToStormValueMapper.java | 69 --------------------
.../hbase/topology/WordCountValueMapper.java | 69 ++++++++++++++++++++
.../storm/hbase/trident/WordCountTrident.java | 6 +-
9 files changed, 123 insertions(+), 123 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index 5cc5ec2..4673024 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
#Storm HBase
-Storm/Trident integration for [Apache HBase]()
+Storm/Trident integration for [Apache HBase](https://hbase.apache.org)
## Usage
The main API for interacting with HBase is the `org.apache.storm.hbase.bolt.mapper.HBaseMapper`
@@ -63,11 +63,11 @@ HBaseBolt hbase = new HBaseBolt("WordCount", mapper);
The `HBaseBolt` will delegate to the `mapper` instance to figure out how to persist tuple data to HBase.
-###HBaseRowToStormValueMapper
+###HBaseValueMapper
This class allows you to transform the HBase lookup result into storm Values that will be emitted by the `HBaseLookupBolt`.
```java
-public interface HBaseRowToStormValueMapper extends Serializable {
+public interface HBaseValueMapper extends Serializable {
public List<Values> toTuples(Result result) throws Exception;
void declareOutputFields(OutputFieldsDeclarer declarer);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java b/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
index 50dbb9d..12263a6 100644
--- a/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
+++ b/src/main/java/org/apache/storm/hbase/bolt/HBaseLookupBolt.java
@@ -26,7 +26,7 @@ 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.HBaseRowToStormValueMapper;
+import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,11 +39,11 @@ import org.slf4j.LoggerFactory;
public class HBaseLookupBolt extends AbstractHBaseBolt {
private static final Logger LOG = LoggerFactory.getLogger(HBaseLookupBolt.class);
- private HBaseRowToStormValueMapper rowToTupleMapper;
+ private HBaseValueMapper rowToTupleMapper;
private HBaseProjectionCriteria projectionCriteria;
- public HBaseLookupBolt(String tableName, HBaseMapper mapper, HBaseRowToStormValueMapper rowToTupleMapper){
+ public HBaseLookupBolt(String tableName, HBaseMapper mapper, HBaseValueMapper rowToTupleMapper){
super(tableName, mapper);
Validate.notNull(rowToTupleMapper, "rowToTupleMapper can not be null");
this.rowToTupleMapper = rowToTupleMapper;
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseRowToStormValueMapper.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseRowToStormValueMapper.java b/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseRowToStormValueMapper.java
deleted file mode 100644
index c1eabdb..0000000
--- a/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseRowToStormValueMapper.java
+++ /dev/null
@@ -1,41 +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.storm.hbase.bolt.mapper;
-
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Values;
-import org.apache.hadoop.hbase.client.Result;
-
-import java.io.Serializable;
-import java.util.List;
-
-public interface HBaseRowToStormValueMapper extends Serializable {
- /**
- *
- * @param result HBase lookup result instance.
- * @return list of values that should be emitted by the lookup bolt.
- * @throws Exception
- */
- public List<Values> toValues(Result result) throws Exception;
-
- /**
- * declares the output fields for the lookup bolt.
- * @param declarer
- */
- void declareOutputFields(OutputFieldsDeclarer declarer);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java b/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java
new file mode 100644
index 0000000..39ce47a
--- /dev/null
+++ b/src/main/java/org/apache/storm/hbase/bolt/mapper/HBaseValueMapper.java
@@ -0,0 +1,41 @@
+/**
+ * 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.storm.hbase.bolt.mapper;
+
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Values;
+import org.apache.hadoop.hbase.client.Result;
+
+import java.io.Serializable;
+import java.util.List;
+
+public interface HBaseValueMapper extends Serializable {
+ /**
+ *
+ * @param result HBase lookup result instance.
+ * @return list of values that should be emitted by the lookup bolt.
+ * @throws Exception
+ */
+ public List<Values> toValues(Result result) throws Exception;
+
+ /**
+ * declares the output fields for the lookup bolt.
+ * @param declarer
+ */
+ void declareOutputFields(OutputFieldsDeclarer declarer);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/main/java/org/apache/storm/hbase/trident/state/HbaseState.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/storm/hbase/trident/state/HbaseState.java b/src/main/java/org/apache/storm/hbase/trident/state/HbaseState.java
index 75b6072..66decf2 100644
--- a/src/main/java/org/apache/storm/hbase/trident/state/HbaseState.java
+++ b/src/main/java/org/apache/storm/hbase/trident/state/HbaseState.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.HBaseRowToStormValueMapper;
+import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
import org.apache.storm.hbase.common.ColumnList;
import org.apache.storm.hbase.common.HBaseClient;
import org.apache.storm.hbase.trident.mapper.TridentHBaseMapper;
@@ -59,7 +59,7 @@ public class HBaseState implements State {
private TridentHBaseMapper mapper;
private Durability durability = Durability.SKIP_WAL;
private HBaseProjectionCriteria projectionCriteria;
- private HBaseRowToStormValueMapper rowToStormValueMapper;
+ private HBaseValueMapper rowToStormValueMapper;
private String configKey;
private String tableName;
@@ -83,7 +83,7 @@ public class HBaseState implements State {
return this;
}
- public Options withRowToStormValueMapper(HBaseRowToStormValueMapper rowToStormValueMapper) {
+ public Options withRowToStormValueMapper(HBaseValueMapper rowToStormValueMapper) {
this.rowToStormValueMapper = rowToStormValueMapper;
return this;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/test/java/org/apache/storm/hbase/topology/LookupWordCount.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/storm/hbase/topology/LookupWordCount.java b/src/test/java/org/apache/storm/hbase/topology/LookupWordCount.java
index ae4fc00..656bce5 100644
--- a/src/test/java/org/apache/storm/hbase/topology/LookupWordCount.java
+++ b/src/test/java/org/apache/storm/hbase/topology/LookupWordCount.java
@@ -51,7 +51,7 @@ public class LookupWordCount {
HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria();
projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"));
- WordCountRowToStormValueMapper rowToTupleMapper = new WordCountRowToStormValueMapper();
+ WordCountValueMapper rowToTupleMapper = new WordCountValueMapper();
HBaseLookupBolt hBaseLookupBolt = new HBaseLookupBolt("WordCount", mapper, rowToTupleMapper)
.withConfigKey("hbase.conf")
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/test/java/org/apache/storm/hbase/topology/WordCountRowToStormValueMapper.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/storm/hbase/topology/WordCountRowToStormValueMapper.java b/src/test/java/org/apache/storm/hbase/topology/WordCountRowToStormValueMapper.java
deleted file mode 100644
index adb61c7..0000000
--- a/src/test/java/org/apache/storm/hbase/topology/WordCountRowToStormValueMapper.java
+++ /dev/null
@@ -1,69 +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.storm.hbase.topology;
-
-
-import backtype.storm.topology.OutputFieldsDeclarer;
-import backtype.storm.tuple.Fields;
-import backtype.storm.tuple.Values;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.storm.hbase.bolt.mapper.HBaseRowToStormValueMapper;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Takes a Hbase result and returns a value list that has a value instance for each column and corresponding value.
- * So if the result from Hbase was
- * <pre>
- * WORD, COUNT
- * apple, 10
- * bannana, 20
- * </pre>
- *
- * this will return
- * <pre>
- * [WORD, apple]
- * [COUNT, 10]
- * [WORD, banana]
- * [COUNT, 20]
- * </pre>
- *
- */
-public class WordCountRowToStormValueMapper implements HBaseRowToStormValueMapper {
-
- @Override
- public List<Values> toValues(Result result) throws Exception {
- List<Values> values = new ArrayList<Values>();
- Cell[] cells = result.rawCells();
- for(Cell cell : cells) {
- Values value = new Values (Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toLong(CellUtil.cloneValue(cell)));
- values.add(value);
- }
- return values;
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("columnName","columnValue"));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/test/java/org/apache/storm/hbase/topology/WordCountValueMapper.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/storm/hbase/topology/WordCountValueMapper.java b/src/test/java/org/apache/storm/hbase/topology/WordCountValueMapper.java
new file mode 100644
index 0000000..dd2ae20
--- /dev/null
+++ b/src/test/java/org/apache/storm/hbase/topology/WordCountValueMapper.java
@@ -0,0 +1,69 @@
+/**
+ * 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.storm.hbase.topology;
+
+
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Takes a Hbase result and returns a value list that has a value instance for each column and corresponding value.
+ * So if the result from Hbase was
+ * <pre>
+ * WORD, COUNT
+ * apple, 10
+ * bannana, 20
+ * </pre>
+ *
+ * this will return
+ * <pre>
+ * [WORD, apple]
+ * [COUNT, 10]
+ * [WORD, banana]
+ * [COUNT, 20]
+ * </pre>
+ *
+ */
+public class WordCountValueMapper implements HBaseValueMapper {
+
+ @Override
+ public List<Values> toValues(Result result) throws Exception {
+ List<Values> values = new ArrayList<Values>();
+ Cell[] cells = result.rawCells();
+ for(Cell cell : cells) {
+ Values value = new Values (Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toLong(CellUtil.cloneValue(cell)));
+ values.add(value);
+ }
+ return values;
+ }
+
+ @Override
+ public void declareOutputFields(OutputFieldsDeclarer declarer) {
+ declarer.declare(new Fields("columnName","columnValue"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/fafda269/src/test/java/org/apache/storm/hbase/trident/WordCountTrident.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/storm/hbase/trident/WordCountTrident.java b/src/test/java/org/apache/storm/hbase/trident/WordCountTrident.java
index 35b3f02..69d067d 100644
--- a/src/test/java/org/apache/storm/hbase/trident/WordCountTrident.java
+++ b/src/test/java/org/apache/storm/hbase/trident/WordCountTrident.java
@@ -25,8 +25,8 @@ import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.storm.hbase.bolt.mapper.HBaseProjectionCriteria;
-import org.apache.storm.hbase.bolt.mapper.HBaseRowToStormValueMapper;
-import org.apache.storm.hbase.topology.WordCountRowToStormValueMapper;
+import org.apache.storm.hbase.bolt.mapper.HBaseValueMapper;
+import org.apache.storm.hbase.topology.WordCountValueMapper;
import org.apache.storm.hbase.trident.mapper.SimpleTridentHBaseMapper;
import org.apache.storm.hbase.trident.mapper.TridentHBaseMapper;
import org.apache.storm.hbase.trident.state.HBaseQuery;
@@ -56,7 +56,7 @@ public class WordCountTrident {
.withCounterFields(new Fields("count"))
.withRowKeyField("word");
- HBaseRowToStormValueMapper rowToStormValueMapper = new WordCountRowToStormValueMapper();
+ HBaseValueMapper rowToStormValueMapper = new WordCountValueMapper();
HBaseProjectionCriteria projectionCriteria = new HBaseProjectionCriteria();
projectionCriteria.addColumn(new HBaseProjectionCriteria.ColumnMetaData("cf", "count"));