View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.mapreduce.replication;
20  
21  import java.io.IOException;
22  
23  import org.apache.commons.logging.Log;
24  import org.apache.commons.logging.LogFactory;
25  import org.apache.hadoop.conf.Configuration;
26  import org.apache.hadoop.conf.Configured;
27  import org.apache.hadoop.hbase.*;
28  import org.apache.hadoop.hbase.client.HConnectable;
29  import org.apache.hadoop.hbase.client.HConnection;
30  import org.apache.hadoop.hbase.client.HConnectionManager;
31  import org.apache.hadoop.hbase.client.HTable;
32  import org.apache.hadoop.hbase.client.Put;
33  import org.apache.hadoop.hbase.client.Result;
34  import org.apache.hadoop.hbase.client.ResultScanner;
35  import org.apache.hadoop.hbase.client.Scan;
36  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
37  import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
38  import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
39  import org.apache.hadoop.hbase.mapreduce.TableMapper;
40  import org.apache.hadoop.hbase.replication.ReplicationException;
41  import org.apache.hadoop.hbase.replication.ReplicationFactory;
42  import org.apache.hadoop.hbase.replication.ReplicationPeerZKImpl;
43  import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
44  import org.apache.hadoop.hbase.replication.ReplicationPeers;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.util.Pair;
47  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
48  import org.apache.hadoop.hbase.zookeeper.ZKUtil;
49  import org.apache.hadoop.mapreduce.Job;
50  import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
51  import org.apache.hadoop.util.Tool;
52  import org.apache.hadoop.util.ToolRunner;
53  
54  /**
55   * This map-only job compares the data from a local table with a remote one.
56   * Every cell is compared and must have exactly the same keys (even timestamp)
57   * as well as same value. It is possible to restrict the job by time range and
58   * families. The peer id that's provided must match the one given when the
59   * replication stream was setup.
60   * <p>
61   * Two counters are provided, Verifier.Counters.GOODROWS and BADROWS. The reason
62   * for a why a row is different is shown in the map's log.
63   */
64  public class VerifyReplication extends Configured implements Tool {
65  
66    private static final Log LOG =
67        LogFactory.getLog(VerifyReplication.class);
68  
69    public final static String NAME = "verifyrep";
70    static long startTime = 0;
71    static long endTime = Long.MAX_VALUE;
72    static int versions = -1;
73    static String tableName = null;
74    static String families = null;
75    static String peerId = null;
76  
77    /**
78     * Map-only comparator for 2 tables
79     */
80    public static class Verifier
81        extends TableMapper<ImmutableBytesWritable, Put> {
82  
83      public static enum Counters {GOODROWS, BADROWS}
84  
85      private ResultScanner replicatedScanner;
86  
87      /**
88       * Map method that compares every scanned row with the equivalent from
89       * a distant cluster.
90       * @param row  The current table row key.
91       * @param value  The columns.
92       * @param context  The current context.
93       * @throws IOException When something is broken with the data.
94       */
95      @Override
96      public void map(ImmutableBytesWritable row, final Result value,
97                      Context context)
98          throws IOException {
99        if (replicatedScanner == null) {
100         Configuration conf = context.getConfiguration();
101         final Scan scan = new Scan();
102         scan.setCaching(conf.getInt(TableInputFormat.SCAN_CACHEDROWS, 1));
103         long startTime = conf.getLong(NAME + ".startTime", 0);
104         long endTime = conf.getLong(NAME + ".endTime", Long.MAX_VALUE);
105         String families = conf.get(NAME + ".families", null);
106         if(families != null) {
107           String[] fams = families.split(",");
108           for(String fam : fams) {
109             scan.addFamily(Bytes.toBytes(fam));
110           }
111         }
112         scan.setTimeRange(startTime, endTime);
113         if (versions >= 0) {
114           scan.setMaxVersions(versions);
115         }
116         HConnectionManager.execute(new HConnectable<Void>(conf) {
117           @Override
118           public Void connect(HConnection conn) throws IOException {
119             String zkClusterKey = conf.get(NAME + ".peerQuorumAddress");
120             Configuration peerConf = HBaseConfiguration.create(conf);
121             ZKUtil.applyClusterKeyToConf(peerConf, zkClusterKey);
122 
123             HTable replicatedTable = new HTable(peerConf, conf.get(NAME + ".tableName"));
124             scan.setStartRow(value.getRow());
125             replicatedScanner = replicatedTable.getScanner(scan);
126             return null;
127           }
128         });
129       }
130       Result res = replicatedScanner.next();
131       try {
132         Result.compareResults(value, res);
133         context.getCounter(Counters.GOODROWS).increment(1);
134       } catch (Exception e) {
135         LOG.warn("Bad row", e);
136         context.getCounter(Counters.BADROWS).increment(1);
137       }
138     }
139 
140     @Override
141     protected void cleanup(Context context) {
142       if (replicatedScanner != null) {
143         replicatedScanner.close();
144         replicatedScanner = null;
145       }
146     }
147   }
148 
149   private static String getPeerQuorumAddress(final Configuration conf) throws IOException {
150     ZooKeeperWatcher localZKW = null;
151     ReplicationPeerZKImpl peer = null;
152     try {
153       localZKW = new ZooKeeperWatcher(conf, "VerifyReplication",
154           new Abortable() {
155             @Override public void abort(String why, Throwable e) {}
156             @Override public boolean isAborted() {return false;}
157           });
158 
159       ReplicationPeers rp = ReplicationFactory.getReplicationPeers(localZKW, conf, localZKW);
160       rp.init();
161 
162       Pair<ReplicationPeerConfig, Configuration> pair = rp.getPeerConf(peerId);
163       if (pair == null) {
164         throw new IOException("Couldn't get peer conf!");
165       }
166       Configuration peerConf = rp.getPeerConf(peerId).getSecond();
167       return ZKUtil.getZooKeeperClusterKey(peerConf);
168     } catch (ReplicationException e) {
169       throw new IOException(
170           "An error occured while trying to connect to the remove peer cluster", e);
171     } finally {
172       if (peer != null) {
173         peer.close();
174       }
175       if (localZKW != null) {
176         localZKW.close();
177       }
178     }
179   }
180 
181   /**
182    * Sets up the actual job.
183    *
184    * @param conf  The current configuration.
185    * @param args  The command line parameters.
186    * @return The newly created job.
187    * @throws java.io.IOException When setting up the job fails.
188    */
189   public static Job createSubmittableJob(Configuration conf, String[] args)
190   throws IOException {
191     if (!doCommandLine(args)) {
192       return null;
193     }
194     if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY,
195         HConstants.REPLICATION_ENABLE_DEFAULT)) {
196       throw new IOException("Replication needs to be enabled to verify it.");
197     }
198     conf.set(NAME+".peerId", peerId);
199     conf.set(NAME+".tableName", tableName);
200     conf.setLong(NAME+".startTime", startTime);
201     conf.setLong(NAME+".endTime", endTime);
202     if (families != null) {
203       conf.set(NAME+".families", families);
204     }
205 
206     String peerQuorumAddress = getPeerQuorumAddress(conf);
207     conf.set(NAME + ".peerQuorumAddress", peerQuorumAddress);
208     LOG.info("Peer Quorum Address: " + peerQuorumAddress);
209 
210     Job job = new Job(conf, NAME + "_" + tableName);
211     job.setJarByClass(VerifyReplication.class);
212 
213     Scan scan = new Scan();
214     scan.setTimeRange(startTime, endTime);
215     if (versions >= 0) {
216       scan.setMaxVersions(versions);
217     }
218     if(families != null) {
219       String[] fams = families.split(",");
220       for(String fam : fams) {
221         scan.addFamily(Bytes.toBytes(fam));
222       }
223     }
224     TableMapReduceUtil.initTableMapperJob(tableName, scan,
225         Verifier.class, null, null, job);
226 
227     // Obtain the auth token from peer cluster
228     TableMapReduceUtil.initCredentialsForCluster(job, peerQuorumAddress);
229 
230     job.setOutputFormatClass(NullOutputFormat.class);
231     job.setNumReduceTasks(0);
232     return job;
233   }
234 
235   private static boolean doCommandLine(final String[] args) {
236     if (args.length < 2) {
237       printUsage(null);
238       return false;
239     }
240     try {
241       for (int i = 0; i < args.length; i++) {
242         String cmd = args[i];
243         if (cmd.equals("-h") || cmd.startsWith("--h")) {
244           printUsage(null);
245           return false;
246         }
247 
248         final String startTimeArgKey = "--starttime=";
249         if (cmd.startsWith(startTimeArgKey)) {
250           startTime = Long.parseLong(cmd.substring(startTimeArgKey.length()));
251           continue;
252         }
253 
254         final String endTimeArgKey = "--endtime=";
255         if (cmd.startsWith(endTimeArgKey)) {
256           endTime = Long.parseLong(cmd.substring(endTimeArgKey.length()));
257           continue;
258         }
259 
260         final String versionsArgKey = "--versions=";
261         if (cmd.startsWith(versionsArgKey)) {
262           versions = Integer.parseInt(cmd.substring(versionsArgKey.length()));
263           continue;
264         }
265 
266         final String familiesArgKey = "--families=";
267         if (cmd.startsWith(familiesArgKey)) {
268           families = cmd.substring(familiesArgKey.length());
269           continue;
270         }
271 
272         if (i == args.length-2) {
273           peerId = cmd;
274         }
275 
276         if (i == args.length-1) {
277           tableName = cmd;
278         }
279       }
280     } catch (Exception e) {
281       e.printStackTrace();
282       printUsage("Can't start because " + e.getMessage());
283       return false;
284     }
285     return true;
286   }
287 
288   /*
289    * @param errorMsg Error message.  Can be null.
290    */
291   private static void printUsage(final String errorMsg) {
292     if (errorMsg != null && errorMsg.length() > 0) {
293       System.err.println("ERROR: " + errorMsg);
294     }
295     System.err.println("Usage: verifyrep [--starttime=X]" +
296         " [--stoptime=Y] [--families=A] <peerid> <tablename>");
297     System.err.println();
298     System.err.println("Options:");
299     System.err.println(" starttime    beginning of the time range");
300     System.err.println("              without endtime means from starttime to forever");
301     System.err.println(" endtime      end of the time range");
302     System.err.println(" versions     number of cell versions to verify");
303     System.err.println(" families     comma-separated list of families to copy");
304     System.err.println();
305     System.err.println("Args:");
306     System.err.println(" peerid       Id of the peer used for verification, must match the one given for replication");
307     System.err.println(" tablename    Name of the table to verify");
308     System.err.println();
309     System.err.println("Examples:");
310     System.err.println(" To verify the data replicated from TestTable for a 1 hour window with peer #5 ");
311     System.err.println(" $ bin/hbase " +
312         "org.apache.hadoop.hbase.mapreduce.replication.VerifyReplication" +
313         " --starttime=1265875194289 --endtime=1265878794289 5 TestTable ");
314   }
315 
316   @Override
317   public int run(String[] args) throws Exception {
318     Configuration conf = this.getConf();
319     Job job = createSubmittableJob(conf, args);
320     if (job != null) {
321       return job.waitForCompletion(true) ? 0 : 1;
322     } 
323     return 1;
324   }
325 
326   /**
327    * Main entry point.
328    *
329    * @param args  The command line parameters.
330    * @throws Exception When running the job fails.
331    */
332   public static void main(String[] args) throws Exception {
333     int res = ToolRunner.run(HBaseConfiguration.create(), new VerifyReplication(), args);
334     System.exit(res);
335   }
336 }