1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
40
41
42 @InterfaceAudience.Public
43 @InterfaceStability.Stable
44 public class RowCounter {
45
46
47 static final String NAME = "rowcounter";
48
49
50
51
52 static class RowCounterMapper
53 extends TableMapper<ImmutableBytesWritable, Result> {
54
55
56 public static enum Counters {ROWS}
57
58
59
60
61
62
63
64
65
66
67
68 @Override
69 public void map(ImmutableBytesWritable row, Result values,
70 Context context)
71 throws IOException {
72
73 context.getCounter(Counters.ROWS).increment(1);
74 }
75 }
76
77
78
79
80
81
82
83
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
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
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
164
165 private static void printUsage(String errorMessage) {
166 System.err.println("ERROR: " + errorMessage);
167 printUsage();
168 }
169
170
171
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
184
185
186
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 }