Source code

001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mapreduce;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertTrue;
022import static org.junit.Assert.fail;
023
024import java.io.ByteArrayOutputStream;
025import java.io.IOException;
026import java.io.PrintStream;
027import java.util.ArrayList;
028import org.apache.hadoop.hbase.HBaseClassTestRule;
029import org.apache.hadoop.hbase.HBaseTestingUtility;
030import org.apache.hadoop.hbase.TableName;
031import org.apache.hadoop.hbase.client.Put;
032import org.apache.hadoop.hbase.client.Table;
033import org.apache.hadoop.hbase.testclassification.LargeTests;
034import org.apache.hadoop.hbase.testclassification.MapReduceTests;
035import org.apache.hadoop.hbase.util.Bytes;
036import org.apache.hadoop.hbase.util.LauncherSecurityManager;
037import org.apache.hadoop.mapreduce.Counter;
038import org.apache.hadoop.mapreduce.Job;
039import org.junit.AfterClass;
040import org.junit.BeforeClass;
041import org.junit.ClassRule;
042import org.junit.Test;
043import org.junit.experimental.categories.Category;
044import org.slf4j.Logger;
045import org.slf4j.LoggerFactory;
046
047/**
048 * Test the rowcounter map reduce job.
049 */
050@Category({MapReduceTests.class, LargeTests.class})
051public class TestRowCounter {
052
053  @ClassRule
054  public static final HBaseClassTestRule CLASS_RULE =
055      HBaseClassTestRule.forClass(TestRowCounter.class);
056
057  private static final Logger LOG = LoggerFactory.getLogger(TestRowCounter.class);
058  private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
059  private final static String TABLE_NAME = "testRowCounter";
060  private final static String TABLE_NAME_TS_RANGE = "testRowCounter_ts_range";
061  private final static String COL_FAM = "col_fam";
062  private final static String COL1 = "c1";
063  private final static String COL2 = "c2";
064  private final static String COMPOSITE_COLUMN = "C:A:A";
065  private final static int TOTAL_ROWS = 10;
066  private final static int ROWS_WITH_ONE_COL = 2;
067
068  /**
069   * @throws java.lang.Exception
070   */
071  @BeforeClass
072  public static void setUpBeforeClass() throws Exception {
073    TEST_UTIL.startMiniCluster();
074    Table table = TEST_UTIL.createTable(TableName.valueOf(TABLE_NAME), Bytes.toBytes(COL_FAM));
075    writeRows(table, TOTAL_ROWS, ROWS_WITH_ONE_COL);
076    table.close();
077  }
078
079  /**
080   * @throws java.lang.Exception
081   */
082  @AfterClass
083  public static void tearDownAfterClass() throws Exception {
084    TEST_UTIL.shutdownMiniCluster();
085  }
086
087  /**
088   * Test a case when no column was specified in command line arguments.
089   *
090   * @throws Exception
091   */
092  @Test
093  public void testRowCounterNoColumn() throws Exception {
094    String[] args = new String[] {
095        TABLE_NAME
096    };
097    runRowCount(args, 10);
098  }
099
100  /**
101   * Test a case when the column specified in command line arguments is
102   * exclusive for few rows.
103   *
104   * @throws Exception
105   */
106  @Test
107  public void testRowCounterExclusiveColumn() throws Exception {
108    String[] args = new String[] {
109        TABLE_NAME, COL_FAM + ":" + COL1
110    };
111    runRowCount(args, 8);
112  }
113
114  /**
115   * Test a case when the column specified in command line arguments is
116   * one for which the qualifier contains colons.
117   *
118   * @throws Exception
119   */
120  @Test
121  public void testRowCounterColumnWithColonInQualifier() throws Exception {
122    String[] args = new String[] {
123        TABLE_NAME, COL_FAM + ":" + COMPOSITE_COLUMN
124    };
125    runRowCount(args, 8);
126  }
127
128  /**
129   * Test a case when the column specified in command line arguments is not part
130   * of first KV for a row.
131   *
132   * @throws Exception
133   */
134  @Test
135  public void testRowCounterHiddenColumn() throws Exception {
136    String[] args = new String[] {
137        TABLE_NAME, COL_FAM + ":" + COL2
138    };
139    runRowCount(args, 10);
140  }
141
142
143  /**
144   * Test a case when the column specified in command line arguments is
145   * exclusive for few rows and also a row range filter is specified
146   *
147   * @throws Exception
148   */
149  @Test
150  public void testRowCounterColumnAndRowRange() throws Exception {
151    String[] args = new String[] {
152            TABLE_NAME, "--range=\\x00rov,\\x00rox", COL_FAM + ":" + COL1
153    };
154    runRowCount(args, 8);
155  }
156
157  /**
158   * Test a case when a range is specified with single range of start-end keys
159   * @throws Exception
160   */
161  @Test
162  public void testRowCounterRowSingleRange() throws Exception {
163    String[] args = new String[] {
164        TABLE_NAME, "--range=\\x00row1,\\x00row3"
165    };
166    runRowCount(args, 2);
167  }
168
169  /**
170   * Test a case when a range is specified with single range with end key only
171   * @throws Exception
172   */
173  @Test
174  public void testRowCounterRowSingleRangeUpperBound() throws Exception {
175    String[] args = new String[] {
176      TABLE_NAME, "--range=,\\x00row3"
177    };
178    runRowCount(args, 3);
179  }
180
181  /**
182   * Test a case when a range is specified with two ranges where one range is with end key only
183   * @throws Exception
184   */
185  @Test
186  public void testRowCounterRowMultiRangeUpperBound() throws Exception {
187    String[] args = new String[] {
188      TABLE_NAME, "--range=,\\x00row3;\\x00row5,\\x00row7"
189    };
190    runRowCount(args, 5);
191  }
192
193  /**
194   * Test a case when a range is specified with multiple ranges of start-end keys
195   * @throws Exception
196   */
197  @Test
198  public void testRowCounterRowMultiRange() throws Exception {
199    String[] args = new String[] {
200        TABLE_NAME, "--range=\\x00row1,\\x00row3;\\x00row5,\\x00row8"
201    };
202    runRowCount(args, 5);
203  }
204
205  /**
206   * Test a case when a range is specified with multiple ranges of start-end keys;
207   * one range is filled, another two are not
208   * @throws Exception
209   */
210  @Test
211  public void testRowCounterRowMultiEmptyRange() throws Exception {
212    String[] args = new String[] {
213        TABLE_NAME, "--range=\\x00row1,\\x00row3;;"
214    };
215    runRowCount(args, 2);
216  }
217
218  @Test
219  public void testRowCounter10kRowRange() throws Exception {
220    String tableName = TABLE_NAME + "10k";
221
222    try (Table table = TEST_UTIL.createTable(
223      TableName.valueOf(tableName), Bytes.toBytes(COL_FAM))) {
224      writeRows(table, 10000, 0);
225    }
226    String[] args = new String[] {
227      tableName, "--range=\\x00row9872,\\x00row9875"
228    };
229    runRowCount(args, 3);
230  }
231
232  /**
233   * Test a case when the timerange is specified with --starttime and --endtime options
234   *
235   * @throws Exception
236   */
237  @Test
238  public void testRowCounterTimeRange() throws Exception {
239    final byte[] family = Bytes.toBytes(COL_FAM);
240    final byte[] col1 = Bytes.toBytes(COL1);
241    Put put1 = new Put(Bytes.toBytes("row_timerange_" + 1));
242    Put put2 = new Put(Bytes.toBytes("row_timerange_" + 2));
243    Put put3 = new Put(Bytes.toBytes("row_timerange_" + 3));
244
245    long ts;
246
247    // clean up content of TABLE_NAME
248    Table table = TEST_UTIL.createTable(TableName.valueOf(TABLE_NAME_TS_RANGE), Bytes.toBytes(COL_FAM));
249
250    ts = System.currentTimeMillis();
251    put1.addColumn(family, col1, ts, Bytes.toBytes("val1"));
252    table.put(put1);
253    Thread.sleep(100);
254
255    ts = System.currentTimeMillis();
256    put2.addColumn(family, col1, ts, Bytes.toBytes("val2"));
257    put3.addColumn(family, col1, ts, Bytes.toBytes("val3"));
258    table.put(put2);
259    table.put(put3);
260    table.close();
261
262    String[] args = new String[] {
263        TABLE_NAME_TS_RANGE, COL_FAM + ":" + COL1,
264        "--starttime=" + 0,
265        "--endtime=" + ts
266    };
267    runRowCount(args, 1);
268
269    args = new String[] {
270        TABLE_NAME_TS_RANGE, COL_FAM + ":" + COL1,
271        "--starttime=" + 0,
272        "--endtime=" + (ts - 10)
273    };
274    runRowCount(args, 1);
275
276    args = new String[] {
277        TABLE_NAME_TS_RANGE, COL_FAM + ":" + COL1,
278        "--starttime=" + ts,
279        "--endtime=" + (ts + 1000)
280    };
281    runRowCount(args, 2);
282
283    args = new String[] {
284        TABLE_NAME_TS_RANGE, COL_FAM + ":" + COL1,
285        "--starttime=" + (ts - 30 * 1000),
286        "--endtime=" + (ts + 30 * 1000),
287    };
288    runRowCount(args, 3);
289  }
290
291  /**
292   * Run the RowCounter map reduce job and verify the row count.
293   *
294   * @param args the command line arguments to be used for rowcounter job.
295   * @param expectedCount the expected row count (result of map reduce job).
296   * @throws Exception
297   */
298  private void runRowCount(String[] args, int expectedCount) throws Exception {
299    Job job = RowCounter.createSubmittableJob(TEST_UTIL.getConfiguration(), args);
300    long start = System.currentTimeMillis();
301    job.waitForCompletion(true);
302    long duration = System.currentTimeMillis() - start;
303    LOG.debug("row count duration (ms): " + duration);
304    assertTrue(job.isSuccessful());
305    Counter counter = job.getCounters().findCounter(RowCounter.RowCounterMapper.Counters.ROWS);
306    assertEquals(expectedCount, counter.getValue());
307  }
308
309  /**
310   * Writes TOTAL_ROWS number of distinct rows in to the table. Few rows have
311   * two columns, Few have one.
312   *
313   * @param table
314   * @throws IOException
315   */
316  private static void writeRows(Table table, int totalRows, int rowsWithOneCol) throws IOException {
317    final byte[] family = Bytes.toBytes(COL_FAM);
318    final byte[] value = Bytes.toBytes("abcd");
319    final byte[] col1 = Bytes.toBytes(COL1);
320    final byte[] col2 = Bytes.toBytes(COL2);
321    final byte[] col3 = Bytes.toBytes(COMPOSITE_COLUMN);
322    ArrayList<Put> rowsUpdate = new ArrayList<>();
323    // write few rows with two columns
324    int i = 0;
325    for (; i < totalRows - rowsWithOneCol; i++) {
326      // Use binary rows values to test for HBASE-15287.
327      byte[] row = Bytes.toBytesBinary("\\x00row" + i);
328      Put put = new Put(row);
329      put.addColumn(family, col1, value);
330      put.addColumn(family, col2, value);
331      put.addColumn(family, col3, value);
332      rowsUpdate.add(put);
333    }
334
335    // write few rows with only one column
336    for (; i < totalRows; i++) {
337      byte[] row = Bytes.toBytes("row" + i);
338      Put put = new Put(row);
339      put.addColumn(family, col2, value);
340      rowsUpdate.add(put);
341    }
342    table.put(rowsUpdate);
343  }
344
345  /**
346   * test main method. Import should print help and call System.exit
347   */
348  @Test
349  public void testImportMain() throws Exception {
350    PrintStream oldPrintStream = System.err;
351    SecurityManager SECURITY_MANAGER = System.getSecurityManager();
352    LauncherSecurityManager newSecurityManager= new LauncherSecurityManager();
353    System.setSecurityManager(newSecurityManager);
354    ByteArrayOutputStream data = new ByteArrayOutputStream();
355    String[] args = {};
356    System.setErr(new PrintStream(data));
357    try {
358      System.setErr(new PrintStream(data));
359
360      try {
361        RowCounter.main(args);
362        fail("should be SecurityException");
363      } catch (SecurityException e) {
364        assertEquals(-1, newSecurityManager.getExitCode());
365        assertTrue(data.toString().contains("Wrong number of parameters:"));
366        assertUsageContent(data.toString());
367      }
368      data.reset();
369      try {
370        args = new String[2];
371        args[0] = "table";
372        args[1] = "--range=1";
373        RowCounter.main(args);
374        fail("should be SecurityException");
375      } catch (SecurityException e) {
376        assertEquals(-1, newSecurityManager.getExitCode());
377        assertTrue(data.toString().contains(
378            "Please specify range in such format as \"--range=a,b\" or, with only one boundary," +
379            " \"--range=,b\" or \"--range=a,\""));
380        assertUsageContent(data.toString());
381      }
382
383    } finally {
384      System.setErr(oldPrintStream);
385      System.setSecurityManager(SECURITY_MANAGER);
386    }
387  }
388
389  private void assertUsageContent(String usage) {
390    assertTrue(usage.contains("Usage: hbase rowcounter [options] <tablename> "
391        + "[--starttime=<start> --endtime=<end>] "
392        + "[--range=[startKey],[endKey][;[startKey],[endKey]...]] [<column1> <column2>...]"));
393    assertTrue(usage.contains("For performance consider the following options:"));
394    assertTrue(usage.contains("-Dhbase.client.scanner.caching=100"));
395    assertTrue(usage.contains("-Dmapreduce.map.speculative=false"));
396  }
397
398}