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 org.apache.hadoop.hbase.classification.InterfaceAudience;
22 import org.apache.hadoop.hbase.classification.InterfaceStability;
23
24 import java.io.IOException;
25 import java.util.HashMap;
26 import java.util.Map;
27 import java.util.Random;
28
29 import org.apache.commons.logging.Log;
30 import org.apache.commons.logging.LogFactory;
31 import org.apache.hadoop.conf.Configuration;
32 import org.apache.hadoop.conf.Configured;
33 import org.apache.hadoop.fs.FileSystem;
34 import org.apache.hadoop.fs.Path;
35 import org.apache.hadoop.hbase.HBaseConfiguration;
36 import org.apache.hadoop.hbase.HConstants;
37 import org.apache.hadoop.hbase.TableName;
38 import org.apache.hadoop.hbase.client.Connection;
39 import org.apache.hadoop.hbase.client.ConnectionFactory;
40 import org.apache.hadoop.hbase.client.Scan;
41 import org.apache.hadoop.hbase.client.Table;
42 import org.apache.hadoop.hbase.util.Bytes;
43 import org.apache.hadoop.mapreduce.Job;
44 import org.apache.hadoop.util.GenericOptionsParser;
45 import org.apache.hadoop.util.Tool;
46 import org.apache.hadoop.util.ToolRunner;
47
48
49
50
51
52
53 @InterfaceAudience.Public
54 @InterfaceStability.Stable
55 public class CopyTable extends Configured implements Tool {
56 private static final Log LOG = LogFactory.getLog(CopyTable.class);
57
58 final static String NAME = "copytable";
59 long startTime = 0;
60 long endTime = 0;
61 int versions = -1;
62 String tableName = null;
63 String startRow = null;
64 String stopRow = null;
65 String dstTableName = null;
66 String peerAddress = null;
67 String families = null;
68 boolean allCells = false;
69 static boolean shuffle = false;
70
71 boolean bulkload = false;
72 Path bulkloadDir = null;
73
74 private final static String JOB_NAME_CONF_KEY = "mapreduce.job.name";
75
76
77
78
79 @Deprecated
80 static long startTime_ = 0;
81 @Deprecated
82 static long endTime_ = 0;
83 @Deprecated
84 static int versions_ = -1;
85 @Deprecated
86 static String tableName_ = null;
87 @Deprecated
88 static String startRow_ = null;
89 @Deprecated
90 static String stopRow_ = null;
91 @Deprecated
92 static String newTableName_ = null;
93 @Deprecated
94 static String peerAddress_ = null;
95 @Deprecated
96 static String families_ = null;
97 @Deprecated
98 static boolean allCells_ = false;
99
100 public CopyTable(Configuration conf) {
101 super(conf);
102 }
103
104
105
106
107
108
109
110
111
112
113 @Deprecated
114 public static Job createSubmittableJob(Configuration conf, String[] args)
115 throws IOException {
116 if (!deprecatedDoCommandLine(args)) {
117 return null;
118 }
119 Job job = new Job(conf, NAME + "_" + tableName_);
120 job.setJarByClass(CopyTable.class);
121 Scan scan = new Scan();
122 scan.setCacheBlocks(false);
123 if (startTime_ != 0) {
124 scan.setTimeRange(startTime_,
125 endTime_ == 0 ? HConstants.LATEST_TIMESTAMP : endTime_);
126 }
127 if (allCells_) {
128 scan.setRaw(true);
129 }
130 if (versions_ >= 0) {
131 scan.setMaxVersions(versions_);
132 }
133 if (startRow_ != null) {
134 scan.setStartRow(Bytes.toBytes(startRow_));
135 }
136 if (stopRow_ != null) {
137 scan.setStopRow(Bytes.toBytes(stopRow_));
138 }
139 if(families_ != null) {
140 String[] fams = families_.split(",");
141 Map<String,String> cfRenameMap = new HashMap<String,String>();
142 for(String fam : fams) {
143 String sourceCf;
144 if(fam.contains(":")) {
145
146 String[] srcAndDest = fam.split(":", 2);
147 sourceCf = srcAndDest[0];
148 String destCf = srcAndDest[1];
149 cfRenameMap.put(sourceCf, destCf);
150 } else {
151
152 sourceCf = fam;
153 }
154 scan.addFamily(Bytes.toBytes(sourceCf));
155 }
156 Import.configureCfRenaming(job.getConfiguration(), cfRenameMap);
157 }
158 TableMapReduceUtil.initTableMapperJob(tableName_, scan,
159 Import.Importer.class, null, null, job);
160 TableMapReduceUtil.initTableReducerJob(
161 newTableName_ == null ? tableName_ : newTableName_, null, job,
162 null, peerAddress_, null, null);
163 job.setNumReduceTasks(0);
164 return job;
165 }
166
167 private static boolean deprecatedDoCommandLine(final String[] args) {
168
169
170 if (args.length < 1) {
171 printUsage(null);
172 return false;
173 }
174 try {
175 for (int i = 0; i < args.length; i++) {
176 String cmd = args[i];
177 if (cmd.equals("-h") || cmd.startsWith("--h")) {
178 printUsage(null);
179 return false;
180 }
181 final String startRowArgKey = "--startrow=";
182 if (cmd.startsWith(startRowArgKey)) {
183 startRow_ = cmd.substring(startRowArgKey.length());
184 continue;
185 }
186 final String stopRowArgKey = "--stoprow=";
187 if (cmd.startsWith(stopRowArgKey)) {
188 stopRow_ = cmd.substring(stopRowArgKey.length());
189 continue;
190 }
191 final String startTimeArgKey = "--starttime=";
192 if (cmd.startsWith(startTimeArgKey)) {
193 startTime_ = Long.parseLong(cmd.substring(startTimeArgKey.length()));
194 continue;
195 }
196 final String endTimeArgKey = "--endtime=";
197 if (cmd.startsWith(endTimeArgKey)) {
198 endTime_ = Long.parseLong(cmd.substring(endTimeArgKey.length()));
199 continue;
200 }
201 final String versionsArgKey = "--versions=";
202 if (cmd.startsWith(versionsArgKey)) {
203 versions_ = Integer.parseInt(cmd.substring(versionsArgKey.length()));
204 continue;
205 }
206 final String newNameArgKey = "--new.name=";
207 if (cmd.startsWith(newNameArgKey)) {
208 newTableName_ = cmd.substring(newNameArgKey.length());
209 continue;
210 }
211 final String peerAdrArgKey = "--peer.adr=";
212 if (cmd.startsWith(peerAdrArgKey)) {
213 peerAddress_ = cmd.substring(peerAdrArgKey.length());
214 continue;
215 }
216 final String familiesArgKey = "--families=";
217 if (cmd.startsWith(familiesArgKey)) {
218 families_ = cmd.substring(familiesArgKey.length());
219 continue;
220 }
221 if (cmd.startsWith("--all.cells")) {
222 allCells_ = true;
223 continue;
224 }
225 if (i == args.length-1) {
226 tableName_ = cmd;
227 } else {
228 printUsage("Invalid argument '" + cmd + "'" );
229 return false;
230 }
231 }
232 if (newTableName_ == null && peerAddress_ == null) {
233 printUsage("At least a new table name or a " +
234 "peer address must be specified");
235 return false;
236 }
237 if ((endTime_ != 0) && (startTime_ > endTime_)) {
238 printUsage("Invalid time range filter: starttime=" + startTime_ + " > endtime="
239 + endTime_);
240 return false;
241 }
242 } catch (Exception e) {
243 e.printStackTrace();
244 printUsage("Can't start because " + e.getMessage());
245 return false;
246 }
247 return true;
248 }
249
250
251
252
253
254
255
256
257 public Job createSubmittableJob(String[] args)
258 throws IOException {
259 if (!doCommandLine(args)) {
260 return null;
261 }
262
263 Job job = Job.getInstance(getConf(), getConf().get(JOB_NAME_CONF_KEY, NAME + "_" + tableName));
264
265 job.setJarByClass(CopyTable.class);
266 Scan scan = new Scan();
267 scan.setCacheBlocks(false);
268 if (startTime != 0) {
269 scan.setTimeRange(startTime,
270 endTime == 0 ? HConstants.LATEST_TIMESTAMP : endTime);
271 }
272 if (allCells) {
273 scan.setRaw(true);
274 }
275 if (shuffle) {
276 job.getConfiguration().set(TableInputFormat.SHUFFLE_MAPS, "true");
277 }
278 if (versions >= 0) {
279 scan.setMaxVersions(versions);
280 }
281
282 if (startRow != null) {
283 scan.setStartRow(Bytes.toBytes(startRow));
284 }
285
286 if (stopRow != null) {
287 scan.setStopRow(Bytes.toBytes(stopRow));
288 }
289
290 if(families != null) {
291 String[] fams = families.split(",");
292 Map<String,String> cfRenameMap = new HashMap<String,String>();
293 for(String fam : fams) {
294 String sourceCf;
295 if(fam.contains(":")) {
296
297 String[] srcAndDest = fam.split(":", 2);
298 sourceCf = srcAndDest[0];
299 String destCf = srcAndDest[1];
300 cfRenameMap.put(sourceCf, destCf);
301 } else {
302
303 sourceCf = fam;
304 }
305 scan.addFamily(Bytes.toBytes(sourceCf));
306 }
307 Import.configureCfRenaming(job.getConfiguration(), cfRenameMap);
308 }
309 job.setNumReduceTasks(0);
310
311 if (bulkload) {
312 TableMapReduceUtil.initTableMapperJob(tableName, scan, Import.KeyValueImporter.class, null,
313 null, job);
314
315
316 TableInputFormat.configureSplitTable(job, TableName.valueOf(dstTableName));
317
318 FileSystem fs = FileSystem.get(getConf());
319 Random rand = new Random();
320 Path root = new Path(fs.getWorkingDirectory(), "copytable");
321 fs.mkdirs(root);
322 while (true) {
323 bulkloadDir = new Path(root, "" + rand.nextLong());
324 if (!fs.exists(bulkloadDir)) {
325 break;
326 }
327 }
328
329 System.out.println("HFiles will be stored at " + this.bulkloadDir);
330 HFileOutputFormat2.setOutputPath(job, bulkloadDir);
331 try (Connection conn = ConnectionFactory.createConnection(getConf());
332 Table htable = conn.getTable(TableName.valueOf(dstTableName))) {
333 HFileOutputFormat2.configureIncrementalLoadMap(job, htable);
334 }
335 } else {
336 TableMapReduceUtil.initTableMapperJob(tableName, scan,
337 Import.Importer.class, null, null, job);
338
339 TableMapReduceUtil.initTableReducerJob(dstTableName, null, job, null, peerAddress, null,
340 null);
341 }
342
343 return job;
344 }
345
346
347
348
349 private static void printUsage(final String errorMsg) {
350 if (errorMsg != null && errorMsg.length() > 0) {
351 System.err.println("ERROR: " + errorMsg);
352 }
353 System.err.println("Usage: CopyTable [general options] [--starttime=X] [--endtime=Y] " +
354 "[--new.name=NEW] [--peer.adr=ADR] <tablename>");
355 System.err.println();
356 System.err.println("Options:");
357 System.err.println(" rs.class hbase.regionserver.class of the peer cluster");
358 System.err.println(" specify if different from current cluster");
359 System.err.println(" rs.impl hbase.regionserver.impl of the peer cluster");
360 System.err.println(" startrow the start row");
361 System.err.println(" stoprow the stop row");
362 System.err.println(" starttime beginning of the time range (unixtime in millis)");
363 System.err.println(" without endtime means from starttime to forever");
364 System.err.println(" endtime end of the time range. Ignored if no starttime specified.");
365 System.err.println(" versions number of cell versions to copy");
366 System.err.println(" new.name new table's name");
367 System.err.println(" peer.adr Address of the peer cluster given in the format");
368 System.err.println(" hbase.zookeeer.quorum:hbase.zookeeper.client.port:zookeeper.znode.parent");
369 System.err.println(" families comma-separated list of families to copy");
370 System.err.println(" To copy from cf1 to cf2, give sourceCfName:destCfName. ");
371 System.err.println(" To keep the same name, just give \"cfName\"");
372 System.err.println(" all.cells also copy delete markers and deleted cells");
373 System.err.println(" bulkload Write input into HFiles and bulk load to the destination "
374 + "table");
375 System.err.println();
376 System.err.println("Args:");
377 System.err.println(" tablename Name of the table to copy");
378 System.err.println();
379 System.err.println("Examples:");
380 System.err.println(" To copy 'TestTable' to a cluster that uses replication for a 1 hour window:");
381 System.err.println(" $ bin/hbase " +
382 "org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1265875194289 --endtime=1265878794289 " +
383 "--peer.adr=server1,server2,server3:2181:/hbase --families=myOldCf:myNewCf,cf2,cf3 TestTable ");
384 System.err.println("For performance consider the following general option:\n"
385 + " It is recommended that you set the following to >=100. A higher value uses more memory but\n"
386 + " decreases the round trip time to the server and may increase performance.\n"
387 + " -Dhbase.client.scanner.caching=100\n"
388 + " The following should always be set to false, to prevent writing data twice, which may produce \n"
389 + " inaccurate results.\n"
390 + " -Dmapreduce.map.speculative=false");
391 }
392
393 private boolean doCommandLine(final String[] args) {
394
395
396 if (args.length < 1) {
397 printUsage(null);
398 return false;
399 }
400 try {
401 for (int i = 0; i < args.length; i++) {
402 String cmd = args[i];
403 if (cmd.equals("-h") || cmd.startsWith("--h")) {
404 printUsage(null);
405 return false;
406 }
407
408 final String startRowArgKey = "--startrow=";
409 if (cmd.startsWith(startRowArgKey)) {
410 startRow = cmd.substring(startRowArgKey.length());
411 continue;
412 }
413
414 final String stopRowArgKey = "--stoprow=";
415 if (cmd.startsWith(stopRowArgKey)) {
416 stopRow = cmd.substring(stopRowArgKey.length());
417 continue;
418 }
419
420 final String startTimeArgKey = "--starttime=";
421 if (cmd.startsWith(startTimeArgKey)) {
422 startTime = Long.parseLong(cmd.substring(startTimeArgKey.length()));
423 continue;
424 }
425
426 final String endTimeArgKey = "--endtime=";
427 if (cmd.startsWith(endTimeArgKey)) {
428 endTime = Long.parseLong(cmd.substring(endTimeArgKey.length()));
429 continue;
430 }
431
432 final String versionsArgKey = "--versions=";
433 if (cmd.startsWith(versionsArgKey)) {
434 versions = Integer.parseInt(cmd.substring(versionsArgKey.length()));
435 continue;
436 }
437
438 final String newNameArgKey = "--new.name=";
439 if (cmd.startsWith(newNameArgKey)) {
440 dstTableName = cmd.substring(newNameArgKey.length());
441 continue;
442 }
443
444 final String peerAdrArgKey = "--peer.adr=";
445 if (cmd.startsWith(peerAdrArgKey)) {
446 peerAddress = cmd.substring(peerAdrArgKey.length());
447 continue;
448 }
449
450 final String familiesArgKey = "--families=";
451 if (cmd.startsWith(familiesArgKey)) {
452 families = cmd.substring(familiesArgKey.length());
453 continue;
454 }
455
456 if (cmd.startsWith("--all.cells")) {
457 allCells = true;
458 continue;
459 }
460
461 if (cmd.startsWith("--bulkload")) {
462 bulkload = true;
463 continue;
464 }
465
466 if (cmd.startsWith("--shuffle")) {
467 shuffle = true;
468 continue;
469 }
470
471 if (i == args.length-1) {
472 tableName = cmd;
473 } else {
474 printUsage("Invalid argument '" + cmd + "'" );
475 return false;
476 }
477 }
478 if (dstTableName == null && peerAddress == null) {
479 printUsage("At least a new table name or a " +
480 "peer address must be specified");
481 return false;
482 }
483 if ((endTime != 0) && (startTime > endTime)) {
484 printUsage("Invalid time range filter: starttime=" + startTime + " > endtime=" + endTime);
485 return false;
486 }
487
488 if (bulkload && peerAddress != null) {
489 printUsage("Remote bulkload is not supported!");
490 return false;
491 }
492
493
494 if (dstTableName == null) {
495 dstTableName = tableName;
496 }
497 } catch (Exception e) {
498 e.printStackTrace();
499 printUsage("Can't start because " + e.getMessage());
500 return false;
501 }
502 return true;
503 }
504
505
506
507
508
509
510
511 public static void main(String[] args) throws Exception {
512 int ret = ToolRunner.run(new CopyTable(HBaseConfiguration.create()), args);
513 System.exit(ret);
514 }
515
516 @Override
517 public int run(String[] args) throws Exception {
518 String[] otherArgs = new GenericOptionsParser(getConf(), args).getRemainingArgs();
519 Job job = createSubmittableJob(otherArgs);
520 if (job == null) return 1;
521 if (!job.waitForCompletion(true)) {
522 LOG.info("Map-reduce job failed!");
523 if (bulkload) {
524 LOG.info("Files are not bulkloaded!");
525 }
526 return 1;
527 }
528 int code = 0;
529 if (bulkload) {
530 code = new LoadIncrementalHFiles(this.getConf()).run(new String[]{this.bulkloadDir.toString(),
531 this.dstTableName});
532 if (code == 0) {
533
534
535 FileSystem fs = FileSystem.get(this.getConf());
536 if (!fs.delete(this.bulkloadDir, true)) {
537 LOG.error("Deleting folder " + bulkloadDir + " failed!");
538 code = 1;
539 }
540 }
541 }
542 return code;
543 }
544 }