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 java.io.IOException; 021import java.text.ParseException; 022import java.text.SimpleDateFormat; 023import java.util.ArrayList; 024import java.util.Collections; 025import java.util.HashSet; 026import java.util.List; 027import java.util.Map; 028import java.util.Set; 029import java.util.TreeMap; 030import org.apache.hadoop.conf.Configuration; 031import org.apache.hadoop.conf.Configured; 032import org.apache.hadoop.fs.Path; 033import org.apache.hadoop.hbase.Cell; 034import org.apache.hadoop.hbase.CellUtil; 035import org.apache.hadoop.hbase.HBaseConfiguration; 036import org.apache.hadoop.hbase.PrivateCellUtil; 037import org.apache.hadoop.hbase.TableName; 038import org.apache.hadoop.hbase.client.Connection; 039import org.apache.hadoop.hbase.client.ConnectionFactory; 040import org.apache.hadoop.hbase.client.Delete; 041import org.apache.hadoop.hbase.client.Mutation; 042import org.apache.hadoop.hbase.client.Put; 043import org.apache.hadoop.hbase.client.RegionLocator; 044import org.apache.hadoop.hbase.client.Table; 045import org.apache.hadoop.hbase.io.ImmutableBytesWritable; 046import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2.TableInfo; 047import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec; 048import org.apache.hadoop.hbase.util.Bytes; 049import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 050import org.apache.hadoop.hbase.util.MapReduceExtendedCell; 051import org.apache.hadoop.hbase.wal.WALEdit; 052import org.apache.hadoop.hbase.wal.WALKey; 053import org.apache.hadoop.mapreduce.Job; 054import org.apache.hadoop.mapreduce.Mapper; 055import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 056import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 057import org.apache.hadoop.util.Tool; 058import org.apache.hadoop.util.ToolRunner; 059import org.apache.yetus.audience.InterfaceAudience; 060import org.slf4j.Logger; 061import org.slf4j.LoggerFactory; 062 063/** 064 * A tool to replay WAL files as a M/R job. The WAL can be replayed for a set of tables or all 065 * tables, and a time range can be provided (in milliseconds). The WAL is filtered to the passed set 066 * of tables and the output can optionally be mapped to another set of tables. WAL replay can also 067 * generate HFiles for later bulk importing, in that case the WAL is replayed for a single table 068 * only. 069 */ 070@InterfaceAudience.Public 071public class WALPlayer extends Configured implements Tool { 072 private static final Logger LOG = LoggerFactory.getLogger(WALPlayer.class); 073 final static String NAME = "WALPlayer"; 074 public final static String BULK_OUTPUT_CONF_KEY = "wal.bulk.output"; 075 public final static String TABLES_KEY = "wal.input.tables"; 076 public final static String TABLE_MAP_KEY = "wal.input.tablesmap"; 077 public final static String INPUT_FILES_SEPARATOR_KEY = "wal.input.separator"; 078 public final static String IGNORE_MISSING_FILES = "wal.input.ignore.missing.files"; 079 public final static String MULTI_TABLES_SUPPORT = "wal.multi.tables.support"; 080 081 protected static final String tableSeparator = ";"; 082 083 private final static String JOB_NAME_CONF_KEY = "mapreduce.job.name"; 084 085 public WALPlayer() { 086 } 087 088 protected WALPlayer(final Configuration c) { 089 super(c); 090 } 091 092 /** 093 * A mapper that just writes out KeyValues. This one can be used together with 094 * {@link CellSortReducer} 095 */ 096 static class WALKeyValueMapper extends Mapper<WALKey, WALEdit, ImmutableBytesWritable, Cell> { 097 private Set<String> tableSet = new HashSet<String>(); 098 private boolean multiTableSupport = false; 099 100 @Override 101 public void map(WALKey key, WALEdit value, Context context) throws IOException { 102 try { 103 // skip all other tables 104 TableName table = key.getTableName(); 105 if (tableSet.contains(table.getNameAsString())) { 106 for (Cell cell : value.getCells()) { 107 if (WALEdit.isMetaEditFamily(cell)) { 108 continue; 109 } 110 111 // Set sequenceId from WALKey, since it is not included by WALCellCodec. The sequenceId 112 // on WALKey is the same value that was on the cells in the WALEdit. This enables 113 // CellSortReducer to use sequenceId to disambiguate duplicate cell timestamps. 114 // See HBASE-27649 115 PrivateCellUtil.setSequenceId(cell, key.getSequenceId()); 116 117 byte[] outKey = multiTableSupport 118 ? Bytes.add(table.getName(), Bytes.toBytes(tableSeparator), CellUtil.cloneRow(cell)) 119 : CellUtil.cloneRow(cell); 120 context.write(new ImmutableBytesWritable(outKey), new MapReduceExtendedCell(cell)); 121 } 122 } 123 } catch (InterruptedException e) { 124 LOG.error("Interrupted while emitting Cell", e); 125 Thread.currentThread().interrupt(); 126 } 127 } 128 129 @Override 130 public void setup(Context context) throws IOException { 131 Configuration conf = context.getConfiguration(); 132 String[] tables = conf.getStrings(TABLES_KEY); 133 this.multiTableSupport = conf.getBoolean(MULTI_TABLES_SUPPORT, false); 134 Collections.addAll(tableSet, tables); 135 } 136 } 137 138 /** 139 * Enum for map metrics. Keep it out here rather than inside in the Map inner-class so we can find 140 * associated properties. 141 */ 142 protected static enum Counter { 143 /** Number of aggregated writes */ 144 PUTS, 145 /** Number of aggregated deletes */ 146 DELETES, 147 CELLS_READ, 148 CELLS_WRITTEN, 149 WALEDITS 150 } 151 152 /** 153 * A mapper that writes out {@link Mutation} to be directly applied to a running HBase instance. 154 */ 155 protected static class WALMapper 156 extends Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation> { 157 private Map<TableName, TableName> tables = new TreeMap<>(); 158 159 @Override 160 public void map(WALKey key, WALEdit value, Context context) throws IOException { 161 context.getCounter(Counter.WALEDITS).increment(1); 162 try { 163 if (tables.isEmpty() || tables.containsKey(key.getTableName())) { 164 TableName targetTable = 165 tables.isEmpty() ? key.getTableName() : tables.get(key.getTableName()); 166 ImmutableBytesWritable tableOut = new ImmutableBytesWritable(targetTable.getName()); 167 Put put = null; 168 Delete del = null; 169 Cell lastCell = null; 170 for (Cell cell : value.getCells()) { 171 context.getCounter(Counter.CELLS_READ).increment(1); 172 // Filtering WAL meta marker entries. 173 if (WALEdit.isMetaEditFamily(cell)) { 174 continue; 175 } 176 // Allow a subclass filter out this cell. 177 if (filter(context, cell)) { 178 // A WALEdit may contain multiple operations (HBASE-3584) and/or 179 // multiple rows (HBASE-5229). 180 // Aggregate as much as possible into a single Put/Delete 181 // operation before writing to the context. 182 if ( 183 lastCell == null || lastCell.getTypeByte() != cell.getTypeByte() 184 || !CellUtil.matchingRows(lastCell, cell) 185 ) { 186 // row or type changed, write out aggregate KVs. 187 if (put != null) { 188 context.write(tableOut, put); 189 context.getCounter(Counter.PUTS).increment(1); 190 } 191 if (del != null) { 192 context.write(tableOut, del); 193 context.getCounter(Counter.DELETES).increment(1); 194 } 195 if (CellUtil.isDelete(cell)) { 196 del = new Delete(CellUtil.cloneRow(cell)); 197 } else { 198 put = new Put(CellUtil.cloneRow(cell)); 199 } 200 } 201 if (CellUtil.isDelete(cell)) { 202 del.add(cell); 203 } else { 204 put.add(cell); 205 } 206 context.getCounter(Counter.CELLS_WRITTEN).increment(1); 207 } 208 lastCell = cell; 209 } 210 // write residual KVs 211 if (put != null) { 212 context.write(tableOut, put); 213 context.getCounter(Counter.PUTS).increment(1); 214 } 215 if (del != null) { 216 context.getCounter(Counter.DELETES).increment(1); 217 context.write(tableOut, del); 218 } 219 } 220 } catch (InterruptedException e) { 221 LOG.error("Interrupted while writing results", e); 222 Thread.currentThread().interrupt(); 223 } 224 } 225 226 protected boolean filter(Context context, final Cell cell) { 227 return true; 228 } 229 230 @Override 231 protected void 232 cleanup(Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation>.Context context) 233 throws IOException, InterruptedException { 234 super.cleanup(context); 235 } 236 237 @SuppressWarnings("checkstyle:EmptyBlock") 238 @Override 239 public void setup(Context context) throws IOException { 240 String[] tableMap = context.getConfiguration().getStrings(TABLE_MAP_KEY); 241 String[] tablesToUse = context.getConfiguration().getStrings(TABLES_KEY); 242 if (tableMap == null) { 243 tableMap = tablesToUse; 244 } 245 if (tablesToUse == null) { 246 // Then user wants all tables. 247 } else if (tablesToUse.length != tableMap.length) { 248 // this can only happen when WALMapper is used directly by a class other than WALPlayer 249 throw new IOException("Incorrect table mapping specified ."); 250 } 251 int i = 0; 252 if (tablesToUse != null) { 253 for (String table : tablesToUse) { 254 tables.put(TableName.valueOf(table), TableName.valueOf(tableMap[i++])); 255 } 256 } 257 } 258 } 259 260 void setupTime(Configuration conf, String option) throws IOException { 261 String val = conf.get(option); 262 if (null == val) { 263 return; 264 } 265 long ms; 266 try { 267 // first try to parse in user friendly form 268 ms = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SS").parse(val).getTime(); 269 } catch (ParseException pe) { 270 try { 271 // then see if just a number of ms's was specified 272 ms = Long.parseLong(val); 273 } catch (NumberFormatException nfe) { 274 throw new IOException( 275 option + " must be specified either in the form 2001-02-20T16:35:06.99 " 276 + "or as number of milliseconds"); 277 } 278 } 279 conf.setLong(option, ms); 280 } 281 282 /** 283 * Sets up the actual job. 284 * @param args The command line parameters. 285 * @return The newly created job. 286 * @throws IOException When setting up the job fails. 287 */ 288 public Job createSubmittableJob(String[] args) throws IOException { 289 Configuration conf = getConf(); 290 setupTime(conf, WALInputFormat.START_TIME_KEY); 291 setupTime(conf, WALInputFormat.END_TIME_KEY); 292 String inputDirs = args[0]; 293 String[] tables = args.length == 1 ? new String[] {} : args[1].split(","); 294 String[] tableMap; 295 if (args.length > 2) { 296 tableMap = args[2].split(","); 297 if (tableMap.length != tables.length) { 298 throw new IOException("The same number of tables and mapping must be provided."); 299 } 300 } else { 301 // if no mapping is specified, map each table to itself 302 tableMap = tables; 303 } 304 conf.setStrings(TABLES_KEY, tables); 305 conf.setStrings(TABLE_MAP_KEY, tableMap); 306 conf.set(FileInputFormat.INPUT_DIR, inputDirs); 307 Job job = Job.getInstance(conf, 308 conf.get(JOB_NAME_CONF_KEY, NAME + "_" + EnvironmentEdgeManager.currentTime())); 309 job.setJarByClass(WALPlayer.class); 310 311 job.setInputFormatClass(WALInputFormat.class); 312 job.setMapOutputKeyClass(ImmutableBytesWritable.class); 313 314 String hfileOutPath = conf.get(BULK_OUTPUT_CONF_KEY); 315 if (hfileOutPath != null) { 316 LOG.debug("add incremental job :" + hfileOutPath + " from " + inputDirs); 317 318 // WALPlayer needs ExtendedCellSerialization so that sequenceId can be propagated when 319 // sorting cells in CellSortReducer 320 job.getConfiguration().setBoolean(HFileOutputFormat2.EXTENDED_CELL_SERIALIZATION_ENABLED_KEY, 321 true); 322 323 // the bulk HFile case 324 List<TableName> tableNames = getTableNameList(tables); 325 326 job.setMapperClass(WALKeyValueMapper.class); 327 job.setReducerClass(CellSortReducer.class); 328 Path outputDir = new Path(hfileOutPath); 329 FileOutputFormat.setOutputPath(job, outputDir); 330 job.setMapOutputValueClass(MapReduceExtendedCell.class); 331 try (Connection conn = ConnectionFactory.createConnection(conf);) { 332 List<TableInfo> tableInfoList = new ArrayList<TableInfo>(); 333 for (TableName tableName : tableNames) { 334 Table table = conn.getTable(tableName); 335 RegionLocator regionLocator = conn.getRegionLocator(tableName); 336 tableInfoList.add(new TableInfo(table.getDescriptor(), regionLocator)); 337 } 338 MultiTableHFileOutputFormat.configureIncrementalLoad(job, tableInfoList); 339 } 340 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(), 341 org.apache.hbase.thirdparty.com.google.common.base.Preconditions.class); 342 } else { 343 // output to live cluster 344 job.setMapperClass(WALMapper.class); 345 job.setOutputFormatClass(MultiTableOutputFormat.class); 346 TableMapReduceUtil.addDependencyJars(job); 347 TableMapReduceUtil.initCredentials(job); 348 // No reducers. 349 job.setNumReduceTasks(0); 350 } 351 String codecCls = WALCellCodec.getWALCellCodecClass(conf).getName(); 352 try { 353 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(), 354 Class.forName(codecCls)); 355 } catch (Exception e) { 356 throw new IOException("Cannot determine wal codec class " + codecCls, e); 357 } 358 return job; 359 } 360 361 private List<TableName> getTableNameList(String[] tables) { 362 List<TableName> list = new ArrayList<TableName>(); 363 for (String name : tables) { 364 list.add(TableName.valueOf(name)); 365 } 366 return list; 367 } 368 369 /** 370 * Print usage 371 * @param errorMsg Error message. Can be null. 372 */ 373 private void usage(final String errorMsg) { 374 if (errorMsg != null && errorMsg.length() > 0) { 375 System.err.println("ERROR: " + errorMsg); 376 } 377 System.err.println("Usage: " + NAME + " [options] <WAL inputdir> [<tables> <tableMappings>]"); 378 System.err.println(" <WAL inputdir> directory of WALs to replay."); 379 System.err.println(" <tables> comma separated list of tables. If no tables specified,"); 380 System.err.println(" all are imported (even hbase:meta if present)."); 381 System.err.println( 382 " <tableMappings> WAL entries can be mapped to a new set of tables by " + "passing"); 383 System.err 384 .println(" <tableMappings>, a comma separated list of target " + "tables."); 385 System.err 386 .println(" If specified, each table in <tables> must have a " + "mapping."); 387 System.err.println("To generate HFiles to bulk load instead of loading HBase directly, pass:"); 388 System.err.println(" -D" + BULK_OUTPUT_CONF_KEY + "=/path/for/output"); 389 System.err.println(" Only one table can be specified, and no mapping allowed!"); 390 System.err.println("To specify a time range, pass:"); 391 System.err.println(" -D" + WALInputFormat.START_TIME_KEY + "=[date|ms]"); 392 System.err.println(" -D" + WALInputFormat.END_TIME_KEY + "=[date|ms]"); 393 System.err.println(" The start and the end date of timerange (inclusive). The dates can be"); 394 System.err 395 .println(" expressed in milliseconds-since-epoch or yyyy-MM-dd'T'HH:mm:ss.SS " + "format."); 396 System.err.println(" E.g. 1234567890120 or 2009-02-13T23:32:30.12"); 397 System.err.println("Other options:"); 398 System.err.println(" -D" + JOB_NAME_CONF_KEY + "=jobName"); 399 System.err.println(" Use the specified mapreduce job name for the wal player"); 400 System.err.println(" -Dwal.input.separator=' '"); 401 System.err.println(" Change WAL filename separator (WAL dir names use default ','.)"); 402 System.err.println("For performance also consider the following options:\n" 403 + " -Dmapreduce.map.speculative=false\n" + " -Dmapreduce.reduce.speculative=false"); 404 } 405 406 /** 407 * Main entry point. 408 * @param args The command line parameters. 409 * @throws Exception When running the job fails. 410 */ 411 public static void main(String[] args) throws Exception { 412 int ret = ToolRunner.run(new WALPlayer(HBaseConfiguration.create()), args); 413 System.exit(ret); 414 } 415 416 @Override 417 public int run(String[] args) throws Exception { 418 if (args.length < 1) { 419 usage("Wrong number of arguments: " + args.length); 420 System.exit(-1); 421 } 422 Job job = createSubmittableJob(args); 423 return job.waitForCompletion(true) ? 0 : 1; 424 } 425}