View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.mapreduce;
19  
20  import static org.junit.Assert.assertEquals;
21  
22  import java.io.IOException;
23  import java.util.Arrays;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.fs.FileSystem;
28  import org.apache.hadoop.fs.Path;
29  import org.apache.hadoop.hbase.CategoryBasedTimeout;
30  import org.apache.hadoop.hbase.Cell;
31  import org.apache.hadoop.hbase.CellUtil;
32  import org.apache.hadoop.hbase.HBaseTestingUtility;
33  import org.apache.hadoop.hbase.TableName;
34  import org.apache.hadoop.hbase.client.HTable;
35  import org.apache.hadoop.hbase.client.Put;
36  import org.apache.hadoop.hbase.client.Result;
37  import org.apache.hadoop.hbase.client.ResultScanner;
38  import org.apache.hadoop.hbase.client.Scan;
39  import org.apache.hadoop.hbase.client.Table;
40  import org.apache.hadoop.hbase.mapreduce.SyncTable.SyncMapper.Counter;
41  import org.apache.hadoop.hbase.testclassification.LargeTests;
42  import org.apache.hadoop.hbase.util.Bytes;
43  import org.apache.hadoop.mapreduce.Counters;
44  import org.junit.AfterClass;
45  import org.junit.Assert;
46  import org.junit.BeforeClass;
47  import org.junit.Rule;
48  import org.junit.Test;
49  import org.junit.experimental.categories.Category;
50  import org.junit.rules.TestRule;
51  
52  import com.google.common.base.Throwables;
53  
54  /**
55   * Basic test for the SyncTable M/R tool
56   */
57  @Category(LargeTests.class)
58  public class TestSyncTable {
59    @Rule public final TestRule timeout = CategoryBasedTimeout.builder().
60        withTimeout(this.getClass()).withLookingForStuckThread(true).build();
61    private static final Log LOG = LogFactory.getLog(TestSyncTable.class);
62    
63    private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();  
64    
65    @BeforeClass
66    public static void beforeClass() throws Exception {
67      TEST_UTIL.startMiniCluster(3);
68      TEST_UTIL.startMiniMapReduceCluster();
69    }
70    
71    @AfterClass
72    public static void afterClass() throws Exception {
73      TEST_UTIL.shutdownMiniMapReduceCluster();
74      TEST_UTIL.shutdownMiniCluster();
75    }
76    
77    private static byte[][] generateSplits(int numRows, int numRegions) {
78      byte[][] splitRows = new byte[numRegions-1][];
79      for (int i = 1; i < numRegions; i++) {
80        splitRows[i-1] = Bytes.toBytes(numRows * i / numRegions);
81      }
82      return splitRows;
83    }
84    
85    @Test
86    public void testSyncTable() throws Exception {
87      String sourceTableName = "testSourceTable";
88      String targetTableName = "testTargetTable";
89      Path testDir = TEST_UTIL.getDataTestDirOnTestFS("testSyncTable");
90      
91      writeTestData(sourceTableName, targetTableName);
92      hashSourceTable(sourceTableName, testDir);
93      Counters syncCounters = syncTables(sourceTableName, targetTableName, testDir);
94      assertEqualTables(90, sourceTableName, targetTableName);
95      
96      assertEquals(60, syncCounters.findCounter(Counter.ROWSWITHDIFFS).getValue());
97      assertEquals(10, syncCounters.findCounter(Counter.SOURCEMISSINGROWS).getValue());
98      assertEquals(10, syncCounters.findCounter(Counter.TARGETMISSINGROWS).getValue());
99      assertEquals(50, syncCounters.findCounter(Counter.SOURCEMISSINGCELLS).getValue());
100     assertEquals(50, syncCounters.findCounter(Counter.TARGETMISSINGCELLS).getValue());
101     assertEquals(20, syncCounters.findCounter(Counter.DIFFERENTCELLVALUES).getValue());
102     
103     TEST_UTIL.deleteTable(sourceTableName);
104     TEST_UTIL.deleteTable(targetTableName);
105     TEST_UTIL.cleanupDataTestDirOnTestFS();
106   }
107 
108   private void assertEqualTables(int expectedRows, String sourceTableName, String targetTableName) 
109       throws Exception {
110     Table sourceTable = TEST_UTIL.getConnection().getTable(TableName.valueOf(sourceTableName));
111     Table targetTable = TEST_UTIL.getConnection().getTable(TableName.valueOf(targetTableName));
112     
113     ResultScanner sourceScanner = sourceTable.getScanner(new Scan());
114     ResultScanner targetScanner = targetTable.getScanner(new Scan());
115     
116     for (int i = 0; i < expectedRows; i++) {
117       Result sourceRow = sourceScanner.next();
118       Result targetRow = targetScanner.next();
119       
120       LOG.debug("SOURCE row: " + (sourceRow == null ? "null" : Bytes.toInt(sourceRow.getRow()))
121           + " cells:" + sourceRow);
122       LOG.debug("TARGET row: " + (targetRow == null ? "null" : Bytes.toInt(targetRow.getRow()))
123           + " cells:" + targetRow);
124       
125       if (sourceRow == null) {
126         Assert.fail("Expected " + expectedRows
127             + " source rows but only found " + i); 
128       }
129       if (targetRow == null) {
130         Assert.fail("Expected " + expectedRows
131             + " target rows but only found " + i); 
132       }
133       Cell[] sourceCells = sourceRow.rawCells();
134       Cell[] targetCells = targetRow.rawCells();
135       if (sourceCells.length != targetCells.length) {
136         LOG.debug("Source cells: " + Arrays.toString(sourceCells));
137         LOG.debug("Target cells: " + Arrays.toString(targetCells));
138         Assert.fail("Row " + Bytes.toInt(sourceRow.getRow())
139             + " has " + sourceCells.length
140             + " cells in source table but " + targetCells.length
141             + " cells in target table");
142       }
143       for (int j = 0; j < sourceCells.length; j++) {
144         Cell sourceCell = sourceCells[j];
145         Cell targetCell = targetCells[j];
146         try {
147           if (!CellUtil.matchingRow(sourceCell, targetCell)) {
148             Assert.fail("Rows don't match");
149           }
150           if (!CellUtil.matchingFamily(sourceCell, targetCell)) {
151             Assert.fail("Families don't match");
152           }
153           if (!CellUtil.matchingQualifier(sourceCell, targetCell)) {
154             Assert.fail("Qualifiers don't match");
155           }
156           if (!CellUtil.matchingTimestamp(sourceCell, targetCell)) {
157             Assert.fail("Timestamps don't match");
158           }
159           if (!CellUtil.matchingValue(sourceCell, targetCell)) {
160             Assert.fail("Values don't match");
161           }
162         } catch (Throwable t) {
163           LOG.debug("Source cell: " + sourceCell + " target cell: " + targetCell);
164           Throwables.propagate(t);
165         }
166       }
167     }
168     Result sourceRow = sourceScanner.next();
169     if (sourceRow != null) {
170       Assert.fail("Source table has more than " + expectedRows
171           + " rows.  Next row: " + Bytes.toInt(sourceRow.getRow()));
172     }
173     Result targetRow = targetScanner.next();
174     if (targetRow != null) {
175       Assert.fail("Target table has more than " + expectedRows
176           + " rows.  Next row: " + Bytes.toInt(targetRow.getRow()));
177     }
178     sourceScanner.close();
179     targetScanner.close();
180     sourceTable.close();
181     targetTable.close();
182   }
183 
184   private Counters syncTables(String sourceTableName, String targetTableName,
185       Path testDir) throws Exception {
186     SyncTable syncTable = new SyncTable(TEST_UTIL.getConfiguration());
187     int code = syncTable.run(new String[] { 
188         testDir.toString(),
189         sourceTableName,
190         targetTableName
191         });
192     assertEquals("sync table job failed", 0, code);
193     
194     LOG.info("Sync tables completed");
195     return syncTable.counters;
196   }
197 
198   private void hashSourceTable(String sourceTableName, Path testDir)
199       throws Exception, IOException {
200     int numHashFiles = 3;
201     long batchSize = 100;  // should be 2 batches per region
202     int scanBatch = 1;
203     HashTable hashTable = new HashTable(TEST_UTIL.getConfiguration());
204     int code = hashTable.run(new String[] { 
205         "--batchsize=" + batchSize,
206         "--numhashfiles=" + numHashFiles,
207         "--scanbatch=" + scanBatch,
208         sourceTableName,
209         testDir.toString()});
210     assertEquals("hash table job failed", 0, code);
211     
212     FileSystem fs = TEST_UTIL.getTestFileSystem();
213     
214     HashTable.TableHash tableHash = HashTable.TableHash.read(fs.getConf(), testDir);
215     assertEquals(sourceTableName, tableHash.tableName);
216     assertEquals(batchSize, tableHash.batchSize);
217     assertEquals(numHashFiles, tableHash.numHashFiles);
218     assertEquals(numHashFiles - 1, tableHash.partitions.size());
219 
220     LOG.info("Hash table completed");
221   }
222 
223   private void writeTestData(String sourceTableName, String targetTableName)
224       throws Exception {
225     final byte[] family = Bytes.toBytes("family");
226     final byte[] column1 = Bytes.toBytes("c1");
227     final byte[] column2 = Bytes.toBytes("c2");
228     final byte[] value1 = Bytes.toBytes("val1");
229     final byte[] value2 = Bytes.toBytes("val2");
230     final byte[] value3 = Bytes.toBytes("val3");
231     
232     int numRows = 100;
233     int sourceRegions = 10;
234     int targetRegions = 6;
235     
236     HTable sourceTable = TEST_UTIL.createTable(TableName.valueOf(sourceTableName),
237         family, generateSplits(numRows, sourceRegions));
238 
239     HTable targetTable = TEST_UTIL.createTable(TableName.valueOf(targetTableName),
240         family, generateSplits(numRows, targetRegions));
241 
242     long timestamp = 1430764183454L;
243 
244     int rowIndex = 0;
245     // a bunch of identical rows
246     for (; rowIndex < 40; rowIndex++) {
247       Put sourcePut = new Put(Bytes.toBytes(rowIndex));
248       sourcePut.addColumn(family, column1, timestamp, value1);
249       sourcePut.addColumn(family, column2, timestamp, value2);
250       sourceTable.put(sourcePut);
251      
252       Put targetPut = new Put(Bytes.toBytes(rowIndex));
253       targetPut.addColumn(family, column1, timestamp, value1);
254       targetPut.addColumn(family, column2, timestamp, value2);
255       targetTable.put(targetPut);
256     }
257     // some rows only in the source table
258     // ROWSWITHDIFFS: 10
259     // TARGETMISSINGROWS: 10
260     // TARGETMISSINGCELLS: 20
261     for (; rowIndex < 50; rowIndex++) {
262       Put put = new Put(Bytes.toBytes(rowIndex));
263       put.addColumn(family, column1, timestamp, value1);
264       put.addColumn(family, column2, timestamp, value2);
265       sourceTable.put(put);
266     }
267     // some rows only in the target table
268     // ROWSWITHDIFFS: 10
269     // SOURCEMISSINGROWS: 10
270     // SOURCEMISSINGCELLS: 20
271     for (; rowIndex < 60; rowIndex++) {
272       Put put = new Put(Bytes.toBytes(rowIndex));
273       put.addColumn(family, column1, timestamp, value1);
274       put.addColumn(family, column2, timestamp, value2);
275       targetTable.put(put);
276     }
277     // some rows with 1 missing cell in target table
278     // ROWSWITHDIFFS: 10
279     // TARGETMISSINGCELLS: 10
280     for (; rowIndex < 70; rowIndex++) {
281       Put sourcePut = new Put(Bytes.toBytes(rowIndex));
282       sourcePut.addColumn(family, column1, timestamp, value1);
283       sourcePut.addColumn(family, column2, timestamp, value2);
284       sourceTable.put(sourcePut);
285 
286       Put targetPut = new Put(Bytes.toBytes(rowIndex));
287       targetPut.addColumn(family, column1, timestamp, value1);
288       targetTable.put(targetPut);
289     }
290     // some rows with 1 missing cell in source table
291     // ROWSWITHDIFFS: 10
292     // SOURCEMISSINGCELLS: 10
293     for (; rowIndex < 80; rowIndex++) {
294       Put sourcePut = new Put(Bytes.toBytes(rowIndex));
295       sourcePut.addColumn(family, column1, timestamp, value1);
296       sourceTable.put(sourcePut);
297 
298       Put targetPut = new Put(Bytes.toBytes(rowIndex));
299       targetPut.addColumn(family, column1, timestamp, value1);
300       targetPut.addColumn(family, column2, timestamp, value2);
301       targetTable.put(targetPut);
302     }
303     // some rows differing only in timestamp
304     // ROWSWITHDIFFS: 10
305     // SOURCEMISSINGCELLS: 20
306     // TARGETMISSINGCELLS: 20
307     for (; rowIndex < 90; rowIndex++) {
308       Put sourcePut = new Put(Bytes.toBytes(rowIndex));
309       sourcePut.addColumn(family, column1, timestamp, column1);
310       sourcePut.addColumn(family, column2, timestamp, value2);
311       sourceTable.put(sourcePut);
312 
313       Put targetPut = new Put(Bytes.toBytes(rowIndex));
314       targetPut.addColumn(family, column1, timestamp+1, column1);
315       targetPut.addColumn(family, column2, timestamp-1, value2);
316       targetTable.put(targetPut);
317     }
318     // some rows with different values
319     // ROWSWITHDIFFS: 10
320     // DIFFERENTCELLVALUES: 20
321     for (; rowIndex < numRows; rowIndex++) {
322       Put sourcePut = new Put(Bytes.toBytes(rowIndex));
323       sourcePut.addColumn(family, column1, timestamp, value1);
324       sourcePut.addColumn(family, column2, timestamp, value2);
325       sourceTable.put(sourcePut);
326       
327       Put targetPut = new Put(Bytes.toBytes(rowIndex));
328       targetPut.addColumn(family, column1, timestamp, value3);
329       targetPut.addColumn(family, column2, timestamp, value3);
330       targetTable.put(targetPut);
331     }
332     
333     sourceTable.close();
334     targetTable.close();
335   }
336   
337 
338 }