You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ec...@apache.org on 2012/11/26 19:02:31 UTC
svn commit: r1413751 [1/3] - in /hbase/trunk: hbase-examples/
hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/
hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/
hbase-examples/src/main/java/org/apache/hadoop/hba...
Author: eclark
Date: Mon Nov 26 18:02:29 2012
New Revision: 1413751
URL: http://svn.apache.org/viewvc?rev=1413751&view=rev
Log:
HBASE-7203 Move example CoProcessor into hbase-examples
Added:
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/ZooKeeperScanPolicyObserver.java
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/generated/
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/generated/BulkDeleteProtos.java
hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/generated/ExampleProtos.java
hbase/trunk/hbase-examples/src/main/protobuf/
hbase/trunk/hbase-examples/src/main/protobuf/BulkDelete.proto
hbase/trunk/hbase-examples/src/main/protobuf/Examples.proto
hbase/trunk/hbase-examples/src/test/
hbase/trunk/hbase-examples/src/test/java/
hbase/trunk/hbase-examples/src/test/java/org/
hbase/trunk/hbase-examples/src/test/java/org/apache/
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/coprocessor/
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/coprocessor/example/
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/coprocessor/example/TestBulkDeleteProtocol.java
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/coprocessor/example/TestRowCountEndpoint.java
hbase/trunk/hbase-examples/src/test/java/org/apache/hadoop/hbase/coprocessor/example/TestZooKeeperScanPolicyObserver.java
Removed:
hbase/trunk/hbase-protocol/src/main/protobuf/BulkDelete.proto
hbase/trunk/hbase-protocol/src/main/protobuf/Examples.proto
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/example/
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/example/
Modified:
hbase/trunk/hbase-examples/pom.xml
Modified: hbase/trunk/hbase-examples/pom.xml
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-examples/pom.xml?rev=1413751&r1=1413750&r2=1413751&view=diff
==============================================================================
--- hbase/trunk/hbase-examples/pom.xml (original)
+++ hbase/trunk/hbase-examples/pom.xml Mon Nov 26 18:02:29 2012
@@ -31,46 +31,39 @@
<description>Examples of HBase usage</description>
<build>
<plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <!-- Currently, there are no examples with tests ported.
- Disable secondPartTestsExecution to prevent surefire
- from failing the build. -->
- <executions>
- <execution>
- <id>secondPartTestsExecution</id>
- <phase>none</phase>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <!-- don't build the test jar - no tests as yet -->
- <configuration>
- <skip>true</skip>
- </configuration>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>${surefire.version}</version>
+ <configuration>
+ <!-- Have to set the groups here because we only do
+ split tests in this package, so groups on live in this module -->
+ <groups>${surefire.firstPartGroups}</groups>
+ </configuration>
+ </plugin>
</plugins>
</build>
<dependencies>
<dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
</dependency>
<dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
</dependency>
- <dependency>
+ <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
@@ -79,4 +72,133 @@
<artifactId>slf4j-log4j12</artifactId>
</dependency>
</dependencies>
+ <profiles>
+ <!-- Profiles for building against different hadoop versions -->
+ <!-- There are a lot of common dependencies used here, should investigate
+if we can combine these profiles somehow -->
+ <!-- profile against Hadoop 1.0.x: This is the default. It has to have the same
+ activation property as the parent Hadoop 1.0.x profile to make sure it gets run at
+ the same time. -->
+ <profile>
+ <id>hadoop-1.0</id>
+ <activation>
+ <property>
+ <name>!hadoop.profile</name>
+ </property>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-test</artifactId>
+ </dependency>
+ </dependencies>
+ </profile>
+ <!--
+ profile for building against Hadoop 2.0.0-alpha. Activate using:
+ mvn -Dhadoop.profile=2.0
+ -->
+ <profile>
+ <id>hadoop-2.0</id>
+ <activation>
+ <property>
+ <name>hadoop.profile</name>
+ <value>2.0</value>
+ </property>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minicluster</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>create-mrapp-generated-classpath</id>
+ <phase>generate-test-resources</phase>
+ <goals>
+ <goal>build-classpath</goal>
+ </goals>
+ <configuration>
+ <!-- needed to run the unit test for DS to generate
+ the required classpath that is required in the env
+ of the launch container in the mini mr/yarn cluster
+ -->
+ <outputFile>${project.build.directory}/test-classes/mrapp-generated-classpath</outputFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <!--
+ profile for building against Hadoop 3.0.x. Activate using:
+ mvn -Dhadoop.profile=3.0
+ -->
+ <profile>
+ <id>hadoop-3.0</id>
+ <activation>
+ <property>
+ <name>hadoop.profile</name>
+ <value>3.0</value>
+ </property>
+ </activation>
+ <properties>
+ <hadoop.version>3.0-SNAPSHOT</hadoop.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minicluster</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>create-mrapp-generated-classpath</id>
+ <phase>generate-test-resources</phase>
+ <goals>
+ <goal>build-classpath</goal>
+ </goals>
+ <configuration>
+ <!-- needed to run the unit test for DS to generate
+ the required classpath that is required in the env
+ of the launch container in the mini mr/yarn cluster
+ -->
+ <outputFile>${project.build.directory}/test-classes/mrapp-generated-classpath</outputFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
Added: hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java?rev=1413751&view=auto
==============================================================================
--- hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java (added)
+++ hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/BulkDeleteEndpoint.java Mon Nov 26 18:02:29 2012
@@ -0,0 +1,290 @@
+/*
+ * 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.hadoop.hbase.coprocessor.example;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.Coprocessor;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.HConstants.OperationStatusCode;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.example.generated.BulkDeleteProtos.BulkDeleteRequest;
+import org.apache.hadoop.hbase.coprocessor.example.generated.BulkDeleteProtos.BulkDeleteResponse;
+import org.apache.hadoop.hbase.coprocessor.example.generated.BulkDeleteProtos.BulkDeleteService;
+import org.apache.hadoop.hbase.coprocessor.example.generated.BulkDeleteProtos.BulkDeleteRequest.DeleteType;
+import org.apache.hadoop.hbase.coprocessor.example.generated.BulkDeleteProtos.BulkDeleteResponse.Builder;
+import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.ResponseConverter;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.OperationStatus;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+
+/**
+ * Defines a protocol to delete data in bulk based on a scan. The scan can be range scan or with
+ * conditions(filters) etc.This can be used to delete rows, column family(s), column qualifier(s)
+ * or version(s) of columns.When delete type is FAMILY or COLUMN, which all family(s) or column(s)
+ * getting deleted will be determined by the Scan. Scan need to select all the families/qualifiers
+ * which need to be deleted.When delete type is VERSION, Which column(s) and version(s) to be
+ * deleted will be determined by the Scan. Scan need to select all the qualifiers and its versions
+ * which needs to be deleted.When a timestamp is passed only one version at that timestamp will be
+ * deleted(even if Scan fetches many versions). When timestamp passed as null, all the versions
+ * which the Scan selects will get deleted.
+ *
+ * </br> Example: <code><pre>
+ * Scan scan = new Scan();
+ * // set scan properties(rowkey range, filters, timerange etc).
+ * HTable ht = ...;
+ * long noOfDeletedRows = 0L;
+ * Batch.Call<BulkDeleteService, BulkDeleteResponse> callable =
+ * new Batch.Call<BulkDeleteService, BulkDeleteResponse>() {
+ * ServerRpcController controller = new ServerRpcController();
+ * BlockingRpcCallback<BulkDeleteResponse> rpcCallback =
+ * new BlockingRpcCallback<BulkDeleteResponse>();
+ *
+ * public BulkDeleteResponse call(BulkDeleteService service) throws IOException {
+ * Builder builder = BulkDeleteRequest.newBuilder();
+ * builder.setScan(ProtobufUtil.toScan(scan));
+ * builder.setDeleteType(DeleteType.VERSION);
+ * builder.setRowBatchSize(rowBatchSize);
+ * // Set optional timestamp if needed
+ * builder.setTimestamp(timeStamp);
+ * service.delete(controller, builder.build(), rpcCallback);
+ * return rpcCallback.get();
+ * }
+ * };
+ * Map<byte[], BulkDeleteResponse> result = ht.coprocessorService(BulkDeleteService.class, scan
+ * .getStartRow(), scan.getStopRow(), callable);
+ * for (BulkDeleteResponse response : result.values()) {
+ * noOfDeletedRows += response.getRowsDeleted();
+ * }
+ * </pre></code>
+ */
+public class BulkDeleteEndpoint extends BulkDeleteService implements CoprocessorService,
+ Coprocessor {
+ private static final String NO_OF_VERSIONS_TO_DELETE = "noOfVersionsToDelete";
+ private static final Log LOG = LogFactory.getLog(BulkDeleteEndpoint.class);
+
+ private RegionCoprocessorEnvironment env;
+
+ @Override
+ public Service getService() {
+ return this;
+ }
+
+ @Override
+ public void delete(RpcController controller, BulkDeleteRequest request,
+ RpcCallback<BulkDeleteResponse> done) {
+ long totalRowsDeleted = 0L;
+ long totalVersionsDeleted = 0L;
+ HRegion region = env.getRegion();
+ int rowBatchSize = request.getRowBatchSize();
+ Long timestamp = null;
+ if (request.hasTimestamp()) {
+ timestamp = request.getTimestamp();
+ }
+ DeleteType deleteType = request.getDeleteType();
+ boolean hasMore = true;
+ RegionScanner scanner = null;
+ try {
+ Scan scan = ProtobufUtil.toScan(request.getScan());
+ if (scan.getFilter() == null && deleteType == DeleteType.ROW) {
+ // What we need is just the rowkeys. So only 1st KV from any row is enough.
+ // Only when it is a row delete, we can apply this filter.
+ // In other types we rely on the scan to know which all columns to be deleted.
+ scan.setFilter(new FirstKeyOnlyFilter());
+ }
+ // Here by assume that the scan is perfect with the appropriate
+ // filter and having necessary column(s).
+ scanner = region.getScanner(scan);
+ while (hasMore) {
+ List<List<KeyValue>> deleteRows = new ArrayList<List<KeyValue>>(rowBatchSize);
+ for (int i = 0; i < rowBatchSize; i++) {
+ List<KeyValue> results = new ArrayList<KeyValue>();
+ hasMore = scanner.next(results);
+ if (results.size() > 0) {
+ deleteRows.add(results);
+ }
+ if (!hasMore) {
+ // There are no more rows.
+ break;
+ }
+ }
+ if (deleteRows.size() > 0) {
+ Pair<Mutation, Integer>[] deleteWithLockArr = new Pair[deleteRows.size()];
+ int i = 0;
+ for (List<KeyValue> deleteRow : deleteRows) {
+ Delete delete = createDeleteMutation(deleteRow, deleteType, timestamp);
+ deleteWithLockArr[i++] = new Pair<Mutation, Integer>(delete, null);
+ }
+ OperationStatus[] opStatus = region.batchMutate(deleteWithLockArr);
+ for (i = 0; i < opStatus.length; i++) {
+ if (opStatus[i].getOperationStatusCode() != OperationStatusCode.SUCCESS) {
+ break;
+ }
+ totalRowsDeleted++;
+ if (deleteType == DeleteType.VERSION) {
+ byte[] versionsDeleted = deleteWithLockArr[i].getFirst().getAttribute(
+ NO_OF_VERSIONS_TO_DELETE);
+ if (versionsDeleted != null) {
+ totalVersionsDeleted += Bytes.toInt(versionsDeleted);
+ }
+ }
+ }
+ }
+ }
+ } catch (IOException ioe) {
+ LOG.error(ioe);
+ // Call ServerRpcController#getFailedOn() to retrieve this IOException at client side.
+ ResponseConverter.setControllerException(controller, ioe);
+ } finally {
+ if (scanner != null) {
+ try {
+ scanner.close();
+ } catch (IOException ioe) {
+ LOG.error(ioe);
+ }
+ }
+ }
+ Builder responseBuilder = BulkDeleteResponse.newBuilder();
+ responseBuilder.setRowsDeleted(totalRowsDeleted);
+ if (deleteType == DeleteType.VERSION) {
+ responseBuilder.setVersionsDeleted(totalVersionsDeleted);
+ }
+ BulkDeleteResponse result = responseBuilder.build();
+ done.run(result);
+ }
+
+ private Delete createDeleteMutation(List<KeyValue> deleteRow, DeleteType deleteType,
+ Long timestamp) {
+ long ts;
+ if (timestamp == null) {
+ ts = HConstants.LATEST_TIMESTAMP;
+ } else {
+ ts = timestamp;
+ }
+ // We just need the rowkey. Get it from 1st KV.
+ byte[] row = deleteRow.get(0).getRow();
+ Delete delete = new Delete(row, ts, null);
+ if (deleteType == DeleteType.FAMILY) {
+ Set<byte[]> families = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
+ for (KeyValue kv : deleteRow) {
+ if (families.add(kv.getFamily())) {
+ delete.deleteFamily(kv.getFamily(), ts);
+ }
+ }
+ } else if (deleteType == DeleteType.COLUMN) {
+ Set<Column> columns = new HashSet<Column>();
+ for (KeyValue kv : deleteRow) {
+ Column column = new Column(kv.getFamily(), kv.getQualifier());
+ if (columns.add(column)) {
+ // Making deleteColumns() calls more than once for the same cf:qualifier is not correct
+ // Every call to deleteColumns() will add a new KV to the familymap which will finally
+ // get written to the memstore as part of delete().
+ delete.deleteColumns(column.family, column.qualifier, ts);
+ }
+ }
+ } else if (deleteType == DeleteType.VERSION) {
+ // When some timestamp was passed to the delete() call only one version of the column (with
+ // given timestamp) will be deleted. If no timestamp passed, it will delete N versions.
+ // How many versions will get deleted depends on the Scan being passed. All the KVs that
+ // the scan fetched will get deleted.
+ int noOfVersionsToDelete = 0;
+ if (timestamp == null) {
+ for (KeyValue kv : deleteRow) {
+ delete.deleteColumn(kv.getFamily(), kv.getQualifier(), kv.getTimestamp());
+ noOfVersionsToDelete++;
+ }
+ } else {
+ Set<Column> columns = new HashSet<Column>();
+ for (KeyValue kv : deleteRow) {
+ Column column = new Column(kv.getFamily(), kv.getQualifier());
+ // Only one version of particular column getting deleted.
+ if (columns.add(column)) {
+ delete.deleteColumn(column.family, column.qualifier, ts);
+ noOfVersionsToDelete++;
+ }
+ }
+ }
+ delete.setAttribute(NO_OF_VERSIONS_TO_DELETE, Bytes.toBytes(noOfVersionsToDelete));
+ }
+ return delete;
+ }
+
+ private static class Column {
+ private byte[] family;
+ private byte[] qualifier;
+
+ public Column(byte[] family, byte[] qualifier) {
+ this.family = family;
+ this.qualifier = qualifier;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof Column)) {
+ return false;
+ }
+ Column column = (Column) other;
+ return Bytes.equals(this.family, column.family)
+ && Bytes.equals(this.qualifier, column.qualifier);
+ }
+
+ @Override
+ public int hashCode() {
+ int h = 31;
+ h = h + 13 * Bytes.hashCode(this.family);
+ h = h + 13 * Bytes.hashCode(this.qualifier);
+ return h;
+ }
+ }
+
+ @Override
+ public void start(CoprocessorEnvironment env) throws IOException {
+ if (env instanceof RegionCoprocessorEnvironment) {
+ this.env = (RegionCoprocessorEnvironment) env;
+ } else {
+ throw new CoprocessorException("Must be loaded on a table region!");
+ }
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment env) throws IOException {
+ // nothing to do
+ }
+}
Added: hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java?rev=1413751&view=auto
==============================================================================
--- hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java (added)
+++ hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/RowCountEndpoint.java Mon Nov 26 18:02:29 2012
@@ -0,0 +1,164 @@
+/*
+ * 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.hadoop.hbase.coprocessor.example;
+
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+import org.apache.hadoop.hbase.Coprocessor;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.example.generated.ExampleProtos;
+import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
+import org.apache.hadoop.hbase.protobuf.ResponseConverter;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Sample coprocessor endpoint exposing a Service interface for counting rows and key values.
+ *
+ * <p>
+ * For the protocol buffer definition of the RowCountService, see the source file located under
+ * hbase-server/src/main/protobuf/Examples.proto.
+ * </p>
+ */
+public class RowCountEndpoint extends ExampleProtos.RowCountService
+ implements Coprocessor, CoprocessorService {
+ private RegionCoprocessorEnvironment env;
+
+ public RowCountEndpoint() {
+ }
+
+ /**
+ * Just returns a reference to this object, which implements the RowCounterService interface.
+ */
+ @Override
+ public Service getService() {
+ return this;
+ }
+
+ /**
+ * Returns a count of the rows in the region where this coprocessor is loaded.
+ */
+ @Override
+ public void getRowCount(RpcController controller, ExampleProtos.CountRequest request,
+ RpcCallback<ExampleProtos.CountResponse> done) {
+ Scan scan = new Scan();
+ scan.setFilter(new FirstKeyOnlyFilter());
+ ExampleProtos.CountResponse response = null;
+ InternalScanner scanner = null;
+ try {
+ scanner = env.getRegion().getScanner(scan);
+ List<KeyValue> results = new ArrayList<KeyValue>();
+ boolean hasMore = false;
+ byte[] lastRow = null;
+ long count = 0;
+ do {
+ hasMore = scanner.next(results);
+ for (KeyValue kv : results) {
+ byte[] currentRow = kv.getRow();
+ if (lastRow == null || !Bytes.equals(lastRow, currentRow)) {
+ lastRow = currentRow;
+ count++;
+ }
+ }
+ results.clear();
+ } while (hasMore);
+
+ response = ExampleProtos.CountResponse.newBuilder()
+ .setCount(count).build();
+ } catch (IOException ioe) {
+ ResponseConverter.setControllerException(controller, ioe);
+ } finally {
+ if (scanner != null) {
+ try {
+ scanner.close();
+ } catch (IOException ignored) {}
+ }
+ }
+ done.run(response);
+ }
+
+ /**
+ * Returns a count of all KeyValues in the region where this coprocessor is loaded.
+ */
+ @Override
+ public void getKeyValueCount(RpcController controller, ExampleProtos.CountRequest request,
+ RpcCallback<ExampleProtos.CountResponse> done) {
+ ExampleProtos.CountResponse response = null;
+ InternalScanner scanner = null;
+ try {
+ scanner = env.getRegion().getScanner(new Scan());
+ List<KeyValue> results = new ArrayList<KeyValue>();
+ boolean hasMore = false;
+ long count = 0;
+ do {
+ hasMore = scanner.next(results);
+ for (KeyValue kv : results) {
+ count++;
+ }
+ results.clear();
+ } while (hasMore);
+
+ response = ExampleProtos.CountResponse.newBuilder()
+ .setCount(count).build();
+ } catch (IOException ioe) {
+ ResponseConverter.setControllerException(controller, ioe);
+ } finally {
+ if (scanner != null) {
+ try {
+ scanner.close();
+ } catch (IOException ignored) {}
+ }
+ }
+ done.run(response);
+ }
+
+ /**
+ * Stores a reference to the coprocessor environment provided by the
+ * {@link org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost} from the region where this
+ * coprocessor is loaded. Since this is a coprocessor endpoint, it always expects to be loaded
+ * on a table region, so always expects this to be an instance of
+ * {@link RegionCoprocessorEnvironment}.
+ * @param env the environment provided by the coprocessor host
+ * @throws IOException if the provided environment is not an instance of
+ * {@code RegionCoprocessorEnvironment}
+ */
+ @Override
+ public void start(CoprocessorEnvironment env) throws IOException {
+ if (env instanceof RegionCoprocessorEnvironment) {
+ this.env = (RegionCoprocessorEnvironment)env;
+ } else {
+ throw new CoprocessorException("Must be loaded on a table region!");
+ }
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment env) throws IOException {
+ // nothing to do
+ }
+}
Added: hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/ZooKeeperScanPolicyObserver.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/ZooKeeperScanPolicyObserver.java?rev=1413751&view=auto
==============================================================================
--- hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/ZooKeeperScanPolicyObserver.java (added)
+++ hbase/trunk/hbase-examples/src/main/java/org/apache/hadoop/hbase/coprocessor/example/ZooKeeperScanPolicyObserver.java Mon Nov 26 18:02:29 2012
@@ -0,0 +1,232 @@
+/*
+ * Copyright The Apache Software Foundation
+ *
+ * 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.hadoop.hbase.coprocessor.example;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.NavigableSet;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.regionserver.HStore.ScanInfo;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
+import org.apache.hadoop.hbase.regionserver.ScanType;
+import org.apache.hadoop.hbase.regionserver.Store;
+import org.apache.hadoop.hbase.regionserver.StoreScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * This is an example showing how a RegionObserver could configured
+ * via ZooKeeper in order to control a Region compaction, flush, and scan policy.
+ *
+ * This also demonstrated the use of shared {@link RegionObserver} state.
+ * See {@link RegionCoprocessorEnvironment#getSharedData()}.
+ *
+ * This would be useful for an incremental backup tool, which would indicate the last
+ * time of a successful backup via ZK and instruct HBase to not delete data that was
+ * inserted since (based on wall clock time).
+ *
+ * This implements org.apache.zookeeper.Watcher directly instead of using
+ * {@link ZooKeeperWatcher}, because RegionObservers come and go and currently
+ * listeners registered with ZooKeeperWatcher cannot be removed.
+ */
+public class ZooKeeperScanPolicyObserver extends BaseRegionObserver {
+ public static String node = "/backup/example/lastbackup";
+ public static String zkkey = "ZK";
+ private static final Log LOG = LogFactory.getLog(ZooKeeperScanPolicyObserver.class);
+
+ /**
+ * Internal watcher that keep "data" up to date asynchronously.
+ */
+ private static class ZKWatcher implements Watcher {
+ private byte[] data = null;
+ private ZooKeeper zk;
+ private volatile boolean needSetup = true;
+ private volatile long lastSetupTry = 0;
+
+ public ZKWatcher(ZooKeeper zk) {
+ this.zk = zk;
+ // trigger the listening
+ getData();
+ }
+
+ /**
+ * Get the maintained data. In case of any ZK exceptions this will retry
+ * establishing the connection (but not more than twice/minute).
+ *
+ * getData is on the critical path, so make sure it is fast unless there is
+ * a problem (network partion, ZK ensemble down, etc)
+ * Make sure at most one (unlucky) thread retries and other threads don't pile up
+ * while that threads tries to recreate the connection.
+ *
+ * @return the last know version of the data
+ */
+ public byte[] getData() {
+ // try at most twice/minute
+ if (needSetup && EnvironmentEdgeManager.currentTimeMillis() > lastSetupTry + 30000) {
+ synchronized (this) {
+ // make sure only one thread tries to reconnect
+ if (needSetup) {
+ needSetup = false;
+ } else {
+ return data;
+ }
+ }
+ // do this without the lock held to avoid threads piling up on this lock,
+ // as it can take a while
+ try {
+ LOG.debug("Connecting to ZK");
+ // record this attempt
+ lastSetupTry = EnvironmentEdgeManager.currentTimeMillis();
+ if (zk.exists(node, false) != null) {
+ data = zk.getData(node, this, null);
+ LOG.debug("Read synchronously: "+(data == null ? "null" : Bytes.toLong(data)));
+ } else {
+ zk.exists(node, this);
+ }
+ } catch (Exception x) {
+ // try again if this fails
+ needSetup = true;
+ }
+ }
+ return data;
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ switch(event.getType()) {
+ case NodeDataChanged:
+ case NodeCreated:
+ try {
+ // get data and re-watch
+ data = zk.getData(node, this, null);
+ LOG.debug("Read asynchronously: "+(data == null ? "null" : Bytes.toLong(data)));
+ } catch (InterruptedException ix) {
+ } catch (KeeperException kx) {
+ needSetup = true;
+ }
+ break;
+
+ case NodeDeleted:
+ try {
+ // just re-watch
+ zk.exists(node, this);
+ data = null;
+ } catch (InterruptedException ix) {
+ } catch (KeeperException kx) {
+ needSetup = true;
+ }
+ break;
+
+ default:
+ // ignore
+ }
+ }
+ }
+
+ @Override
+ public void start(CoprocessorEnvironment e) throws IOException {
+ RegionCoprocessorEnvironment re = (RegionCoprocessorEnvironment) e;
+ if (!re.getSharedData().containsKey(zkkey)) {
+ // there is a short race here
+ // in the worst case we create a watcher that will be notified once
+ re.getSharedData().putIfAbsent(
+ zkkey,
+ new ZKWatcher(re.getRegionServerServices().getZooKeeper()
+ .getRecoverableZooKeeper().getZooKeeper()));
+ }
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment e) throws IOException {
+ // nothing to do here
+ }
+
+ protected ScanInfo getScanInfo(HStore store, RegionCoprocessorEnvironment e) {
+ byte[] data = ((ZKWatcher)e.getSharedData().get(zkkey)).getData();
+ if (data == null) {
+ return null;
+ }
+ ScanInfo oldSI = store.getScanInfo();
+ if (oldSI.getTtl() == Long.MAX_VALUE) {
+ return null;
+ }
+ long ttl = Math.max(EnvironmentEdgeManager.currentTimeMillis() - Bytes.toLong(data), oldSI.getTtl());
+ return new ScanInfo(store.getFamily(), ttl,
+ oldSI.getTimeToPurgeDeletes(), oldSI.getComparator());
+ }
+
+ @Override
+ public InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
+ HStore store, KeyValueScanner memstoreScanner, InternalScanner s) throws IOException {
+ HStore.ScanInfo scanInfo = getScanInfo(store, c.getEnvironment());
+ if (scanInfo == null) {
+ // take default action
+ return null;
+ }
+ Scan scan = new Scan();
+ scan.setMaxVersions(scanInfo.getMaxVersions());
+ return new StoreScanner(store, scanInfo, scan, Collections.singletonList(memstoreScanner),
+ ScanType.MINOR_COMPACT, store.getHRegion().getSmallestReadPoint(),
+ HConstants.OLDEST_TIMESTAMP);
+ }
+
+ @Override
+ public InternalScanner preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
+ HStore store, List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs,
+ InternalScanner s) throws IOException {
+ HStore.ScanInfo scanInfo = getScanInfo(store, c.getEnvironment());
+ if (scanInfo == null) {
+ // take default action
+ return null;
+ }
+ Scan scan = new Scan();
+ scan.setMaxVersions(scanInfo.getMaxVersions());
+ return new StoreScanner(store, scanInfo, scan, scanners, scanType, store.getHRegion()
+ .getSmallestReadPoint(), earliestPutTs);
+ }
+
+ @Override
+ public KeyValueScanner preStoreScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
+ final HStore store, final Scan scan, final NavigableSet<byte[]> targetCols,
+ final KeyValueScanner s) throws IOException {
+ HStore.ScanInfo scanInfo = getScanInfo(store, c.getEnvironment());
+ if (scanInfo == null) {
+ // take default action
+ return null;
+ }
+ return new StoreScanner(store, scanInfo, scan, targetCols);
+ }
+}