View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.mapreduce;
20  
21  import java.io.IOException;
22  
23  import org.apache.commons.lang.StringUtils;
24  import org.apache.hadoop.hbase.HConstants;
25  import org.apache.hadoop.hbase.classification.InterfaceAudience;
26  import org.apache.hadoop.hbase.classification.InterfaceStability;
27  import org.apache.hadoop.conf.Configuration;
28  import org.apache.hadoop.hbase.HBaseConfiguration;
29  import org.apache.hadoop.hbase.client.Result;
30  import org.apache.hadoop.hbase.client.Scan;
31  import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
32  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
33  import org.apache.hadoop.hbase.util.Bytes;
34  import org.apache.hadoop.mapreduce.Job;
35  import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
36  import org.apache.hadoop.util.GenericOptionsParser;
37  
38  /**
39   * A job with a just a map phase to count rows. Map outputs table rows IF the
40   * input row has columns that have content.
41   */
42  @InterfaceAudience.Public
43  @InterfaceStability.Stable
44  public class RowCounter {
45  
46    /** Name of this 'program'. */
47    static final String NAME = "rowcounter";
48  
49    /**
50     * Mapper that runs the count.
51     */
52    static class RowCounterMapper
53    extends TableMapper<ImmutableBytesWritable, Result> {
54  
55      /** Counter enumeration to count the actual rows. */
56      public static enum Counters {ROWS}
57  
58      /**
59       * Maps the data.
60       *
61       * @param row  The current table row key.
62       * @param values  The columns.
63       * @param context  The current context.
64       * @throws IOException When something is broken with the data.
65       * @see org.apache.hadoop.mapreduce.Mapper#map(KEYIN, VALUEIN,
66       *   org.apache.hadoop.mapreduce.Mapper.Context)
67       */
68      @Override
69      public void map(ImmutableBytesWritable row, Result values,
70        Context context)
71      throws IOException {
72        // Count every row containing data, whether it's in qualifiers or values
73        context.getCounter(Counters.ROWS).increment(1);
74      }
75    }
76  
77    /**
78     * Sets up the actual job.
79     *
80     * @param conf  The current configuration.
81     * @param args  The command line parameters.
82     * @return The newly created job.
83     * @throws IOException When setting up the job fails.
84     */
85    public static Job createSubmittableJob(Configuration conf, String[] args)
86    throws IOException {
87      String tableName = args[0];
88      String startKey = null;
89      String endKey = null;
90      long startTime = 0;
91      long endTime = 0;
92  
93      StringBuilder sb = new StringBuilder();
94  
95      final String rangeSwitch = "--range=";
96      final String startTimeArgKey = "--starttime=";
97      final String endTimeArgKey = "--endtime=";
98  
99      // First argument is table name, starting from second
100     for (int i = 1; i < args.length; i++) {
101       if (args[i].startsWith(rangeSwitch)) {
102         String[] startEnd = args[i].substring(rangeSwitch.length()).split(",", 2);
103         if (startEnd.length != 2 || startEnd[1].contains(",")) {
104           printUsage("Please specify range in such format as \"--range=a,b\" " +
105               "or, with only one boundary, \"--range=,b\" or \"--range=a,\"");
106           return null;
107         }
108         startKey = startEnd[0];
109         endKey = startEnd[1];
110       }
111       if (startTime < endTime) {
112         printUsage("--endtime=" + endTime + " needs to be greater than --starttime=" + startTime);
113         return null;
114       }
115       if (args[i].startsWith(startTimeArgKey)) {
116         startTime = Long.parseLong(args[i].substring(startTimeArgKey.length()));
117         continue;
118       }
119       if (args[i].startsWith(endTimeArgKey)) {
120         endTime = Long.parseLong(args[i].substring(endTimeArgKey.length()));
121         continue;
122       }
123       else {
124         // if no switch, assume column names
125         sb.append(args[i]);
126         sb.append(" ");
127       }
128     }
129 
130     Job job = new Job(conf, NAME + "_" + tableName);
131     job.setJarByClass(RowCounter.class);
132     Scan scan = new Scan();
133     scan.setCacheBlocks(false);
134     if (startKey != null && !startKey.equals("")) {
135       scan.setStartRow(Bytes.toBytesBinary(startKey));
136     }
137     if (endKey != null && !endKey.equals("")) {
138       scan.setStopRow(Bytes.toBytesBinary(endKey));
139     }
140     if (sb.length() > 0) {
141       for (String columnName : sb.toString().trim().split(" ")) {
142         String family = StringUtils.substringBefore(columnName, ":");
143         String qualifier = StringUtils.substringAfter(columnName, ":");
144 
145         if (StringUtils.isBlank(qualifier)) {
146           scan.addFamily(Bytes.toBytes(family));
147         }
148         else {
149           scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
150         }
151       }
152     }
153     scan.setFilter(new FirstKeyOnlyFilter()); 
154     scan.setTimeRange(startTime, endTime == 0 ? HConstants.LATEST_TIMESTAMP : endTime);
155     job.setOutputFormatClass(NullOutputFormat.class);
156     TableMapReduceUtil.initTableMapperJob(tableName, scan,
157       RowCounterMapper.class, ImmutableBytesWritable.class, Result.class, job);
158     job.setNumReduceTasks(0);
159     return job;
160   }
161 
162   /*
163    * @param errorMessage Can attach a message when error occurs.
164    */
165   private static void printUsage(String errorMessage) {
166     System.err.println("ERROR: " + errorMessage);
167     printUsage();
168   }
169 
170   /*
171    * Prints usage without error message
172    */
173   private static void printUsage() {
174     System.err.println("Usage: RowCounter [options] <tablename> " +
175         "[--starttime=[start] --endtime=[end] " +
176         "[--range=[startKey],[endKey]] [<column1> <column2>...]");
177     System.err.println("For performance consider the following options:\n"
178         + "-Dhbase.client.scanner.caching=100\n"
179         + "-Dmapreduce.map.speculative=false");
180   }
181 
182   /**
183    * Main entry point.
184    *
185    * @param args  The command line parameters.
186    * @throws Exception When running the job fails.
187    */
188   public static void main(String[] args) throws Exception {
189     Configuration conf = HBaseConfiguration.create();
190     String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
191     if (otherArgs.length < 1) {
192       printUsage("Wrong number of parameters: " + args.length);
193       System.exit(-1);
194     }
195     Job job = createSubmittableJob(conf, otherArgs);
196     if (job == null) {
197       System.exit(-1);
198     }
199     System.exit(job.waitForCompletion(true) ? 0 : 1);
200   }
201 }