1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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;
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
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
258
259
260
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
268
269
270
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
278
279
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
291
292
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
304
305
306
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
319
320
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 }