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 */
018
019package org.apache.hadoop.hbase.test;
020
021import java.io.BufferedReader;
022import java.io.DataInput;
023import java.io.DataOutput;
024import java.io.IOException;
025import java.io.InputStreamReader;
026import java.net.URI;
027import java.net.URISyntaxException;
028import java.nio.charset.StandardCharsets;
029import java.util.ArrayList;
030import java.util.HashSet;
031import java.util.List;
032import java.util.Set;
033import java.util.concurrent.atomic.AtomicLong;
034import java.util.zip.GZIPInputStream;
035import org.apache.hadoop.conf.Configuration;
036import org.apache.hadoop.conf.Configured;
037import org.apache.hadoop.fs.FSDataInputStream;
038import org.apache.hadoop.fs.FileSystem;
039import org.apache.hadoop.fs.Path;
040import org.apache.hadoop.hbase.Cell;
041import org.apache.hadoop.hbase.HBaseConfiguration;
042import org.apache.hadoop.hbase.HBaseTestingUtil;
043import org.apache.hadoop.hbase.HConstants;
044import org.apache.hadoop.hbase.IntegrationTestBase;
045import org.apache.hadoop.hbase.IntegrationTestingUtility;
046import org.apache.hadoop.hbase.MasterNotRunningException;
047import org.apache.hadoop.hbase.TableName;
048import org.apache.hadoop.hbase.client.Admin;
049import org.apache.hadoop.hbase.client.BufferedMutator;
050import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
051import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
052import org.apache.hadoop.hbase.client.Connection;
053import org.apache.hadoop.hbase.client.ConnectionFactory;
054import org.apache.hadoop.hbase.client.Get;
055import org.apache.hadoop.hbase.client.Put;
056import org.apache.hadoop.hbase.client.Result;
057import org.apache.hadoop.hbase.client.Table;
058import org.apache.hadoop.hbase.client.TableDescriptor;
059import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
060import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
061import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
062import org.apache.hadoop.hbase.regionserver.BloomType;
063import org.apache.hadoop.hbase.test.util.CRC64;
064import org.apache.hadoop.hbase.test.util.warc.WARCInputFormat;
065import org.apache.hadoop.hbase.test.util.warc.WARCRecord;
066import org.apache.hadoop.hbase.test.util.warc.WARCWritable;
067import org.apache.hadoop.hbase.util.Bytes;
068import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
069import org.apache.hadoop.hbase.util.RegionSplitter;
070import org.apache.hadoop.io.BytesWritable;
071import org.apache.hadoop.io.LongWritable;
072import org.apache.hadoop.io.NullWritable;
073import org.apache.hadoop.io.SequenceFile.CompressionType;
074import org.apache.hadoop.io.Writable;
075import org.apache.hadoop.mapreduce.Counters;
076import org.apache.hadoop.mapreduce.Job;
077import org.apache.hadoop.mapreduce.JobContext;
078import org.apache.hadoop.mapreduce.Mapper;
079import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
080import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
081import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
082import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
083import org.apache.hadoop.util.Tool;
084import org.apache.hadoop.util.ToolRunner;
085import org.slf4j.Logger;
086import org.slf4j.LoggerFactory;
087import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
088
089/**
090 * This integration test loads successful resource retrieval records from the Common Crawl
091 * (https://commoncrawl.org/) public dataset into an HBase table and writes records that can be
092 * used to later verify the presence and integrity of those records.
093 * <p>
094 * Run like:
095 * <blockquote>
096 * ./bin/hbase org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl \<br>
097 * &nbsp;&nbsp; -Dfs.s3n.awsAccessKeyId=&lt;AWS access key&gt; \<br>
098 * &nbsp;&nbsp; -Dfs.s3n.awsSecretAccessKey=&lt;AWS secret key&gt; \<br>
099 * &nbsp;&nbsp; /path/to/test-CC-MAIN-2021-10-warc.paths.gz \<br>
100 * &nbsp;&nbsp; /path/to/tmp/warc-loader-output
101 * </blockquote>
102 * <p>
103 * Access to the Common Crawl dataset in S3 is made available to anyone by Amazon AWS, but
104 * Hadoop's S3N filesystem still requires valid access credentials to initialize.
105 * <p>
106 * The input path can either specify a directory or a file. The file may optionally be
107 * compressed with gzip. If a directory, the loader expects the directory to contain one or more
108 * WARC files from the Common Crawl dataset. If a file, the loader expects a list of Hadoop S3N
109 * URIs which point to S3 locations for one or more WARC files from the Common Crawl dataset,
110 * one URI per line. Lines should be terminated with the UNIX line terminator.
111 * <p>
112 * Included in hbase-it/src/test/resources/CC-MAIN-2021-10-warc.paths.gz is a list of all WARC
113 * files comprising the Q1 2021 crawl archive. There are 64,000 WARC files in this data set, each
114 * containing ~1GB of gzipped data. The WARC files contain several record types, such as metadata,
115 * request, and response, but we only load the response record types. If the HBase table schema
116 * does not specify compression (by default) there is roughly a 10x expansion. Loading the full
117 * crawl archive results in a table approximately 640 TB in size.
118 * <p>
119 * You can also split the Loader and Verify stages:
120 * <p>
121 * Load with:
122 * <blockquote>
123 * ./bin/hbase 'org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl$Loader' \<br>
124 * &nbsp;&nbsp; -files /path/to/hadoop-aws.jar \<br>
125 * &nbsp;&nbsp; -Dfs.s3n.awsAccessKeyId=&lt;AWS access key&gt; \<br>
126 * &nbsp;&nbsp; -Dfs.s3n.awsSecretAccessKey=&lt;AWS secret key&gt; \<br>
127 * &nbsp;&nbsp; /path/to/test-CC-MAIN-2021-10-warc.paths.gz \<br>
128 * &nbsp;&nbsp; /path/to/tmp/warc-loader-output
129 * </blockquote>
130 * <p>
131 * Note: The hadoop-aws jar will be needed at runtime to instantiate the S3N filesystem. Use
132 * the <tt>-files</tt> ToolRunner argument to add it.
133 * <p>
134 * Verify with:
135 * <blockquote>
136 * ./bin/hbase 'org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl$Verify' \<br>
137 * &nbsp;&nbsp; /path/to/tmp/warc-loader-output
138 * </blockquote>
139 * <p>
140 */
141public class IntegrationTestLoadCommonCrawl extends IntegrationTestBase {
142
143  private static final Logger LOG = LoggerFactory.getLogger(IntegrationTestLoadCommonCrawl.class);
144
145  protected static String TABLE_NAME_KEY = "IntegrationTestLoadCommonCrawl.table";
146  protected static String DEFAULT_TABLE_NAME = "IntegrationTestLoadCommonCrawl";
147
148  protected static byte[] CONTENT_FAMILY_NAME = Bytes.toBytes("c");
149  protected static byte[] INFO_FAMILY_NAME = Bytes.toBytes("i");
150  protected static byte[] CONTENT_QUALIFIER = HConstants.EMPTY_BYTE_ARRAY;
151  protected static byte[] CONTENT_LENGTH_QUALIFIER = Bytes.toBytes("l");
152  protected static byte[] CONTENT_TYPE_QUALIFIER = Bytes.toBytes("t");
153  protected static byte[] CRC_QUALIFIER = Bytes.toBytes("c");
154  protected static byte[] DATE_QUALIFIER = Bytes.toBytes("d");
155  protected static byte[] IP_ADDRESS_QUALIFIER = Bytes.toBytes("a");
156  protected static byte[] RECORD_ID_QUALIFIER = Bytes.toBytes("r");
157  protected static byte[] TARGET_URI_QUALIFIER = Bytes.toBytes("u");
158
159  private static final int VERIFICATION_READ_RETRIES = 10;
160
161  public static enum Counts {
162    REFERENCED, UNREFERENCED, CORRUPT
163  }
164
165  protected Path warcFileInputDir = null;
166  protected Path outputDir = null;
167  protected String[] args;
168
169  protected int runLoader(final Path warcFileInputDir, final Path outputDir) throws Exception {
170    Loader loader = new Loader();
171    loader.setConf(conf);
172    return loader.run(warcFileInputDir, outputDir);
173  }
174
175  protected int runVerify(final Path inputDir) throws Exception {
176    Verify verify = new Verify();
177    verify.setConf(conf);
178    return verify.run(inputDir);
179  }
180
181  @Override
182  public int run(String[] args) {
183    if (args.length > 0) {
184      warcFileInputDir = new Path(args[0]);
185      if (args.length > 1) {
186        outputDir = new Path(args[1]);
187      }
188    }
189    try {
190      if (warcFileInputDir == null) {
191        throw new IllegalArgumentException("WARC input file or directory not specified");
192      }
193      if (outputDir == null) {
194        throw new IllegalArgumentException("Output directory not specified");
195      }
196      int res = runLoader(warcFileInputDir, outputDir);
197      if (res != 0) {
198        LOG.error("Loader failed");
199        return -1;
200      }
201      res = runVerify(outputDir);
202    } catch (Exception e) {
203      LOG.error("Tool failed with exception", e);
204      return -1;
205    }
206    return 0;
207  }
208
209  @Override
210  protected void processOptions(final CommandLine cmd) {
211    processBaseOptions(cmd);
212    args = cmd.getArgs();
213  }
214
215  @Override
216  public void setUpCluster() throws Exception {
217    util = getTestingUtil(getConf());
218    boolean isDistributed = util.isDistributedCluster();
219    util.initializeCluster(isDistributed ? 1 : 3);
220    if (!isDistributed) {
221      util.startMiniMapReduceCluster();
222    }
223    this.setConf(util.getConfiguration());
224  }
225
226  @Override
227  public void cleanUpCluster() throws Exception {
228    super.cleanUpCluster();
229    if (util.isDistributedCluster()) {
230      util.shutdownMiniMapReduceCluster();
231    }
232  }
233
234  static TableName getTablename(final Configuration c) {
235    return TableName.valueOf(c.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
236  }
237
238  @Override
239  public TableName getTablename() {
240    return getTablename(getConf());
241  }
242
243  @Override
244  protected Set<String> getColumnFamilies() {
245    Set<String> families = new HashSet<>();
246    families.add(Bytes.toString(CONTENT_FAMILY_NAME));
247    families.add(Bytes.toString(INFO_FAMILY_NAME));
248    return families;
249  }
250
251  @Override
252  public int runTestFromCommandLine() throws Exception {
253    return ToolRunner.run(getConf(), this, args);
254  }
255
256  public static void main(String[] args) throws Exception {
257    Configuration conf = HBaseConfiguration.create();
258    IntegrationTestingUtility.setUseDistributedCluster(conf);
259    int ret = ToolRunner.run(conf, new IntegrationTestLoadCommonCrawl(), args);
260    System.exit(ret);
261  }
262
263  public static class HBaseKeyWritable implements Writable {
264
265    private byte[] row;
266    private int rowOffset;
267    private int rowLength;
268    private byte[] family;
269    private int familyOffset;
270    private int familyLength;
271    private byte[] qualifier;
272    private int qualifierOffset;
273    private int qualifierLength;
274    private long ts;
275
276    public HBaseKeyWritable() { }
277
278    public HBaseKeyWritable(byte[] row, int rowOffset, int rowLength,
279        byte[] family, int familyOffset, int familyLength,
280        byte[] qualifier, int qualifierOffset, int qualifierLength, long ts) {
281      this.row = row;
282      this.rowOffset = rowOffset;
283      this.rowLength = rowLength;
284      this.family = family;
285      this.familyOffset = familyOffset;
286      this.familyLength = familyLength;
287      this.qualifier = qualifier;
288      this.qualifierOffset = qualifierOffset;
289      this.qualifierLength = qualifierLength;
290      this.ts = ts;
291    }
292
293    public HBaseKeyWritable(byte[] row, byte[] family, byte[] qualifier, long ts) {
294      this(row, 0, row.length,
295        family, 0, family.length,
296        qualifier, 0, qualifier != null ? qualifier.length : 0, ts);
297    }
298
299    public HBaseKeyWritable(byte[] row, byte[] family, long ts) {
300      this(row, family, HConstants.EMPTY_BYTE_ARRAY, ts);
301    }
302
303    public HBaseKeyWritable(Cell cell) {
304      this(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
305        cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
306        cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength(),
307        cell.getTimestamp());
308    }
309
310    @Override
311    public void readFields(DataInput in) throws IOException {
312      this.row = Bytes.toBytes(in.readUTF());
313      this.rowOffset = 0;
314      this.rowLength = row.length;
315      this.family = Bytes.toBytes(in.readUTF());
316      this.familyOffset = 0;
317      this.familyLength = family.length;
318      this.qualifier = Bytes.toBytes(in.readUTF());
319      this.qualifierOffset = 0;
320      this.qualifierLength = qualifier.length;
321      this.ts = in.readLong();
322    }
323
324    @Override
325    public void write(DataOutput out) throws IOException {
326      out.writeUTF(new String(row, rowOffset, rowLength, StandardCharsets.UTF_8));
327      out.writeUTF(new String(family, familyOffset, familyLength, StandardCharsets.UTF_8));
328      if (qualifier != null) {
329        out.writeUTF(new String(qualifier, qualifierOffset, qualifierLength,
330          StandardCharsets.UTF_8));
331      } else {
332        out.writeUTF("");
333      }
334      out.writeLong(ts);
335    }
336
337    public byte[] getRowArray() {
338      return row;
339    }
340
341    public void setRow(byte[] row) {
342      this.row = row;
343    }
344
345    public int getRowOffset() {
346      return rowOffset;
347    }
348
349    public void setRowOffset(int rowOffset) {
350      this.rowOffset = rowOffset;
351    }
352
353    public int getRowLength() {
354      return rowLength;
355    }
356
357    public void setRowLength(int rowLength) {
358      this.rowLength = rowLength;
359    }
360
361    public byte[] getFamilyArray() {
362      return family;
363    }
364
365    public void setFamily(byte[] family) {
366      this.family = family;
367    }
368
369    public int getFamilyOffset() {
370      return familyOffset;
371    }
372
373    public void setFamilyOffset(int familyOffset) {
374      this.familyOffset = familyOffset;
375    }
376
377    public int getFamilyLength() {
378      return familyLength;
379    }
380
381    public void setFamilyLength(int familyLength) {
382      this.familyLength = familyLength;
383    }
384
385    public byte[] getQualifierArray() {
386      return qualifier;
387    }
388
389    public void setQualifier(byte[] qualifier) {
390      this.qualifier = qualifier;
391    }
392
393    public int getQualifierOffset() {
394      return qualifierOffset;
395    }
396
397    public void setQualifierOffset(int qualifierOffset) {
398      this.qualifierOffset = qualifierOffset;
399    }
400
401    public int getQualifierLength() {
402      return qualifierLength;
403    }
404
405    public void setQualifierLength(int qualifierLength) {
406      this.qualifierLength = qualifierLength;
407    }
408
409    public long getTimestamp() {
410      return ts;
411    }
412
413    public void setTimestamp(long ts) {
414      this.ts = ts;
415    }
416  }
417
418  public static class Loader extends Configured implements Tool {
419
420    private static final Logger LOG = LoggerFactory.getLogger(Loader.class);
421    private static final String USAGE = "Loader <warInputDir | warFileList> <outputDir>";
422
423    void createSchema(final TableName tableName) throws IOException {
424
425      try (Connection conn = ConnectionFactory.createConnection(getConf());
426           Admin admin = conn.getAdmin()) {
427        if (!admin.tableExists(tableName)) {
428
429          ColumnFamilyDescriptorBuilder contentFamilyBuilder =
430            ColumnFamilyDescriptorBuilder.newBuilder(CONTENT_FAMILY_NAME)
431              .setDataBlockEncoding(DataBlockEncoding.NONE)
432              .setBloomFilterType(BloomType.ROW)
433              .setMaxVersions(1000)
434              .setBlocksize(256 * 1024)
435              ;
436
437          ColumnFamilyDescriptorBuilder infoFamilyBuilder =
438            ColumnFamilyDescriptorBuilder.newBuilder(INFO_FAMILY_NAME)
439              .setDataBlockEncoding(DataBlockEncoding.NONE)
440              .setBloomFilterType(BloomType.ROWCOL)
441              .setMaxVersions(1000)
442              .setBlocksize(8 * 1024)
443              ;
444
445          Set<ColumnFamilyDescriptor> families = new HashSet<>();
446          families.add(contentFamilyBuilder.build());
447          families.add(infoFamilyBuilder.build());
448
449          TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
450              .setColumnFamilies(families)
451              .build();
452
453          if (getConf().getBoolean(HBaseTestingUtil.PRESPLIT_TEST_TABLE_KEY,
454            HBaseTestingUtil.PRESPLIT_TEST_TABLE)) {
455            int numberOfServers = admin.getRegionServers().size();
456            if (numberOfServers == 0) {
457              throw new IllegalStateException("No live regionservers");
458            }
459            int regionsPerServer = getConf().getInt(HBaseTestingUtil.REGIONS_PER_SERVER_KEY,
460              HBaseTestingUtil.DEFAULT_REGIONS_PER_SERVER);
461            int totalNumberOfRegions = numberOfServers * regionsPerServer;
462            LOG.info("Creating test table: " + tableDescriptor);
463            LOG.info("Number of live regionservers: " + numberOfServers + ", " +
464                "pre-splitting table into " + totalNumberOfRegions + " regions " +
465                "(default regions per server: " + regionsPerServer + ")");
466            byte[][] splits = new RegionSplitter.UniformSplit().split(totalNumberOfRegions);
467            admin.createTable(tableDescriptor, splits);
468          } else {
469            LOG.info("Creating test table: " + tableDescriptor);
470            admin.createTable(tableDescriptor);
471          }
472        }
473      } catch (MasterNotRunningException e) {
474        LOG.error("Master not running", e);
475        throw new IOException(e);
476      }
477    }
478
479    int run(final Path warcFileInput, final Path outputDir)
480        throws IOException, ClassNotFoundException, InterruptedException {
481
482      createSchema(getTablename(getConf()));
483
484      final Job job = Job.getInstance(getConf());
485      job.setJobName(Loader.class.getName());
486      job.setNumReduceTasks(0);
487      job.setJarByClass(getClass());
488      job.setMapperClass(LoaderMapper.class);
489      job.setInputFormatClass(WARCInputFormat.class);
490      final FileSystem fs = FileSystem.get(warcFileInput.toUri(), getConf());
491      if (fs.getFileStatus(warcFileInput).isDirectory()) {
492        LOG.info("Using directory as WARC input path: " + warcFileInput);
493        FileInputFormat.setInputPaths(job, warcFileInput);
494      } else if (warcFileInput.toUri().getScheme().equals("file")) {
495        LOG.info("Getting WARC input paths from file: " + warcFileInput);
496        final List<Path> paths = new ArrayList<Path>();
497        try (FSDataInputStream is = fs.open(warcFileInput)) {
498          InputStreamReader reader;
499          if (warcFileInput.getName().toLowerCase().endsWith(".gz")) {
500            reader = new InputStreamReader(new GZIPInputStream(is));
501          } else {
502            reader = new InputStreamReader(is);
503          }
504          try (BufferedReader br = new BufferedReader(reader)) {
505            String line;
506            while ((line = br.readLine()) != null) {
507              paths.add(new Path(line));
508            }
509          }
510        }
511        LOG.info("Read " + paths.size() + " WARC input paths from " + warcFileInput);
512        FileInputFormat.setInputPaths(job, paths.toArray(new Path[paths.size()]));
513      } else {
514        FileInputFormat.setInputPaths(job, warcFileInput);
515      }
516      job.setOutputFormatClass(SequenceFileOutputFormat.class);
517      SequenceFileOutputFormat.setOutputPath(job, outputDir);
518      SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);
519      job.setOutputKeyClass(HBaseKeyWritable.class);
520      job.setOutputValueClass(BytesWritable.class);
521      TableMapReduceUtil.addDependencyJars(job);
522
523      boolean success = job.waitForCompletion(true);
524      if (!success) {
525        LOG.error("Failure during job " + job.getJobID());
526      }
527      return success ? 0 : 1;
528    }
529
530    @Override
531    public int run(String[] args) throws Exception {
532      if (args.length < 2) {
533        System.err.println(USAGE);
534        return 1;
535      }
536      try {
537        Path warcFileInput = new Path(args[0]);
538        Path outputDir = new Path(args[1]);
539        return run(warcFileInput, outputDir);
540      } catch (NumberFormatException e) {
541        System.err.println("Parsing loader arguments failed: " + e.getMessage());
542        System.err.println(USAGE);
543        return 1;
544      }
545    }
546
547    public static void main(String[] args) throws Exception {
548      System.exit(ToolRunner.run(HBaseConfiguration.create(), new Loader(), args));
549    }
550
551    public static class LoaderMapper
552        extends Mapper<LongWritable, WARCWritable, HBaseKeyWritable, BytesWritable> {
553
554      protected Configuration conf;
555      protected Connection conn;
556      protected BufferedMutator mutator;
557
558      @Override
559      protected void setup(final Context context) throws IOException, InterruptedException {
560        conf = context.getConfiguration();
561        conn = ConnectionFactory.createConnection(conf);
562        mutator = conn.getBufferedMutator(getTablename(conf));
563      }
564
565      @Override
566      protected void cleanup(final Context context) throws IOException, InterruptedException {
567        try {
568          mutator.close();
569        } catch (Exception e) {
570          LOG.warn("Exception closing Table", e);
571        }
572        try {
573          conn.close();
574        } catch (Exception e) {
575          LOG.warn("Exception closing Connection", e);
576        }
577      }
578
579      @Override
580      protected void map(final LongWritable key, final WARCWritable value, final Context output)
581          throws IOException, InterruptedException {
582        final WARCRecord.Header warcHeader = value.getRecord().getHeader();
583        final String recordID = warcHeader.getRecordID();
584        final String targetURI = warcHeader.getTargetURI();
585        if (warcHeader.getRecordType().equals("response") && targetURI != null) {
586          final String contentType = warcHeader.getField("WARC-Identified-Payload-Type");
587          if (contentType != null) {
588            LOG.info("Processing uri=\"" + targetURI + "\", id=" + recordID);
589
590            // Make row key
591
592            byte[] rowKey;
593            try {
594              rowKey = rowKeyFromTargetURI(targetURI);
595            } catch (IllegalArgumentException e) {
596              LOG.debug("Could not make a row key for record " + recordID + ", ignoring", e);
597              return;
598            } catch (URISyntaxException e) {
599              LOG.warn("Could not parse URI \"" + targetURI + "\" for record " + recordID +
600                ", ignoring");
601              return;
602            }
603
604            // Get the content and calculate the CRC64
605
606            final byte[] content = value.getRecord().getContent();
607            final CRC64 crc = new CRC64();
608            crc.update(content);
609            final long crc64 = crc.getValue();
610
611            // Store to HBase
612
613            final long ts = getCurrentTime();
614            final Put put = new Put(rowKey);
615            put.addColumn(CONTENT_FAMILY_NAME, CONTENT_QUALIFIER, ts, content);
616            put.addColumn(INFO_FAMILY_NAME, CONTENT_LENGTH_QUALIFIER, ts,
617              Bytes.toBytes(content.length));
618            put.addColumn(INFO_FAMILY_NAME, CONTENT_TYPE_QUALIFIER, ts,
619              Bytes.toBytes(contentType));
620            put.addColumn(INFO_FAMILY_NAME, CRC_QUALIFIER, ts, Bytes.toBytes(crc64));
621            put.addColumn(INFO_FAMILY_NAME, RECORD_ID_QUALIFIER, ts, Bytes.toBytes(recordID));
622            put.addColumn(INFO_FAMILY_NAME, TARGET_URI_QUALIFIER, ts, Bytes.toBytes(targetURI));
623            put.addColumn(INFO_FAMILY_NAME, DATE_QUALIFIER, ts,
624              Bytes.toBytes(warcHeader.getDateString()));
625            final String ipAddr = warcHeader.getField("WARC-IP-Address");
626            if (ipAddr != null) {
627              put.addColumn(INFO_FAMILY_NAME, IP_ADDRESS_QUALIFIER, ts, Bytes.toBytes(ipAddr));
628            }
629            mutator.mutate(put);
630
631            // Write records out for later verification, one per HBase field except for the
632            // content record, which will be verified by CRC64.
633
634            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, CRC_QUALIFIER, ts),
635              new BytesWritable(Bytes.toBytes(crc64)));
636            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, CONTENT_LENGTH_QUALIFIER,
637              ts), new BytesWritable(Bytes.toBytes(content.length)));
638            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, CONTENT_TYPE_QUALIFIER,
639              ts), new BytesWritable(Bytes.toBytes(contentType)));
640            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, RECORD_ID_QUALIFIER,
641              ts), new BytesWritable(Bytes.toBytes(recordID)));
642            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, TARGET_URI_QUALIFIER,
643              ts), new BytesWritable(Bytes.toBytes(targetURI)));
644            output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, DATE_QUALIFIER, ts),
645              new BytesWritable(Bytes.toBytes(warcHeader.getDateString())));
646            if (ipAddr != null) {
647              output.write(new HBaseKeyWritable(rowKey, INFO_FAMILY_NAME, IP_ADDRESS_QUALIFIER,
648                ts), new BytesWritable(Bytes.toBytes(ipAddr)));
649            }
650          }
651        }
652      }
653
654      private byte[] rowKeyFromTargetURI(final String targetUri)
655          throws URISyntaxException, IllegalArgumentException {
656        final URI uri = new URI(targetUri);
657        // Ignore the scheme
658        // Reverse the components of the hostname
659        String reversedHost;
660        if (uri.getHost() != null) {
661          final StringBuilder sb = new StringBuilder();
662          final String[] hostComponents = uri.getHost().split("\\.");
663          for (int i = hostComponents.length - 1; i >= 0; i--) {
664            sb.append(hostComponents[i]);
665            if (i != 0) {
666              sb.append('.');
667            }
668          }
669          reversedHost = sb.toString();
670        } else {
671          throw new IllegalArgumentException("URI is missing host component");
672        }
673        final StringBuilder sb = new StringBuilder();
674        sb.append(reversedHost);
675        if (uri.getPort() >= 0) {
676          sb.append(':');
677          sb.append(uri.getPort());
678        }
679        if (uri.getPath() != null) {
680          sb.append('/');
681          sb.append(uri.getPath());
682        }
683        if (uri.getQuery() != null) {
684          sb.append('?');
685          sb.append(uri.getQuery());
686        }
687        if (uri.getFragment() != null) {
688          sb.append('#');
689          sb.append(uri.getFragment());
690        }
691        if (sb.length() > HConstants.MAX_ROW_LENGTH) {
692          throw new IllegalArgumentException("Key would be too large (length=" + sb.length() +
693            ", limit=" + HConstants.MAX_ROW_LENGTH);
694        }
695        return Bytes.toBytes(sb.toString());
696      }
697
698    }
699  }
700
701  public static class OneFilePerMapperSFIF<K, V> extends SequenceFileInputFormat<K, V> {
702    @Override
703    protected boolean isSplitable(final JobContext context, final Path filename) {
704      return false;
705    }
706  }
707
708  public static class Verify extends Configured implements Tool {
709
710    public static final Logger LOG = LoggerFactory.getLogger(Verify.class);
711    public static final String USAGE = "Verify <inputDir>";
712
713    int run(final Path inputDir)
714        throws IOException, ClassNotFoundException, InterruptedException {
715      Job job = Job.getInstance(getConf());
716      job.setJobName(Verify.class.getName());
717      job.setJarByClass(getClass());
718      job.setMapperClass(VerifyMapper.class);
719      job.setInputFormatClass(OneFilePerMapperSFIF.class);
720      FileInputFormat.setInputPaths(job, inputDir);
721      job.setOutputFormatClass(NullOutputFormat.class);
722      job.setOutputKeyClass(NullWritable.class);
723      job.setOutputValueClass(NullWritable.class);
724      TableMapReduceUtil.addDependencyJars(job);
725      boolean success = job.waitForCompletion(true);
726      if (!success) {
727        LOG.error("Failure during job " + job.getJobID());
728      }
729      final Counters counters = job.getCounters();
730      for (Counts c: Counts.values()) {
731        LOG.info(c + ": " + counters.findCounter(c).getValue());
732      }
733      if (counters.findCounter(Counts.UNREFERENCED).getValue() > 0) {
734        LOG.error("Nonzero UNREFERENCED count from job " + job.getJobID());
735        success = false;
736      }
737      if (counters.findCounter(Counts.CORRUPT).getValue() > 0) {
738        LOG.error("Nonzero CORRUPT count from job " + job.getJobID());
739        success = false;
740      }
741      return success ? 0 : 1;
742    }
743
744    @Override
745    public int run(String[] args) throws Exception {
746      if (args.length < 2) {
747        System.err.println(USAGE);
748        return 1;
749      }
750      Path loaderOutput = new Path(args[0]);
751      return run(loaderOutput);
752    }
753
754    public static void main(String[] args) throws Exception {
755      System.exit(ToolRunner.run(HBaseConfiguration.create(), new Verify(), args));
756    }
757
758    public static class VerifyMapper
759        extends Mapper<HBaseKeyWritable, BytesWritable, NullWritable, NullWritable> {
760
761      private Connection conn;
762      private Table table;
763
764      @Override
765      protected void setup(final Context context) throws IOException, InterruptedException {
766        conn = ConnectionFactory.createConnection(context.getConfiguration());
767        table = conn.getTable(getTablename(conn.getConfiguration()));
768      }
769
770      @Override
771      protected void cleanup(final Context context) throws IOException, InterruptedException {
772        try {
773          table.close();
774        } catch (Exception e) {
775          LOG.warn("Exception closing Table", e);
776        }
777        try {
778          conn.close();
779        } catch (Exception e) {
780          LOG.warn("Exception closing Connection", e);
781        }
782      }
783
784      @Override
785      protected void map(final HBaseKeyWritable key, final BytesWritable value,
786          final Context output) throws IOException, InterruptedException {
787        final byte[] row = Bytes.copy(key.getRowArray(), key.getRowOffset(), key.getRowLength());
788        final byte[] family = Bytes.copy(key.getFamilyArray(), key.getFamilyOffset(),
789          key.getFamilyLength());
790        final byte[] qualifier = Bytes.copy(key.getQualifierArray(), key.getQualifierOffset(),
791          key.getQualifierLength());
792        final long ts = key.getTimestamp();
793
794        int retries = VERIFICATION_READ_RETRIES;
795        while (true) {
796
797          if (Bytes.equals(INFO_FAMILY_NAME, family) &&
798              Bytes.equals(CRC_QUALIFIER, qualifier)) {
799
800            final long expectedCRC64 = Bytes.toLong(value.getBytes(), 0, value.getLength());
801            final Result result = table.get(new Get(row)
802              .addColumn(CONTENT_FAMILY_NAME, CONTENT_QUALIFIER)
803              .addColumn(INFO_FAMILY_NAME, CRC_QUALIFIER)
804              .setTimestamp(ts));
805            final byte[] content = result.getValue(CONTENT_FAMILY_NAME, CONTENT_QUALIFIER);
806            if (content == null) {
807              if (retries-- > 0) {
808                continue;
809              }
810              LOG.error("Row " + Bytes.toStringBinary(row) + ": missing content");
811              output.getCounter(Counts.UNREFERENCED).increment(1);
812              return;
813            } else {
814              final CRC64 crc = new CRC64();
815              crc.update(content);
816              if (crc.getValue() != expectedCRC64) {
817                LOG.error("Row " + Bytes.toStringBinary(row) + ": corrupt content");
818                output.getCounter(Counts.CORRUPT).increment(1);
819                return;
820              }
821            }
822            final byte[] crc = result.getValue(INFO_FAMILY_NAME, CRC_QUALIFIER);
823            if (crc == null) {
824              if (retries-- > 0) {
825                continue;
826              }
827              LOG.error("Row " + Bytes.toStringBinary(row) + ": missing i:c");
828              output.getCounter(Counts.UNREFERENCED).increment(1);
829              return;
830            }
831            if (Bytes.toLong(crc) != expectedCRC64) {
832              if (retries-- > 0) {
833                continue;
834              }
835              LOG.error("Row " + Bytes.toStringBinary(row) + ": i:c mismatch");
836              output.getCounter(Counts.CORRUPT).increment(1);
837              return;
838            }
839
840          } else {
841
842            final Result result = table.get(new Get(row)
843              .addColumn(family, qualifier)
844              .setTimestamp(ts));
845            final byte[] bytes = result.getValue(family, qualifier);
846            if (bytes == null) {
847              if (retries-- > 0) {
848                continue;
849              }
850              LOG.error("Row " + Bytes.toStringBinary(row) + ": missing " +
851                Bytes.toStringBinary(family) + ":" + Bytes.toStringBinary(qualifier));
852              output.getCounter(Counts.UNREFERENCED).increment(1);
853              return;
854            }
855            if (!Bytes.equals(bytes, 0, bytes.length, value.getBytes(), 0, value.getLength())) {
856              if (retries-- > 0) {
857                continue;
858              }
859              LOG.error("Row " + Bytes.toStringBinary(row) + ": " +
860                Bytes.toStringBinary(family) + ":" + Bytes.toStringBinary(qualifier) +
861                " mismatch");
862              output.getCounter(Counts.CORRUPT).increment(1);
863              return;
864            }
865
866          }
867
868          // If we fell through to here all verification checks have succeeded, potentially after
869          // retries, and we must exit the while loop.
870          output.getCounter(Counts.REFERENCED).increment(1);
871          break;
872
873        }
874      }
875    }
876  }
877
878  private static final AtomicLong counter = new AtomicLong();
879
880  private static long getCurrentTime() {
881    // Typical hybrid logical clock scheme.
882    // Take the current time, shift by 16 bits and zero those bits, and replace those bits
883    // with the low 16 bits of the atomic counter. Mask off the high bit too because timestamps
884    // cannot be negative.
885    return ((EnvironmentEdgeManager.currentTime() << 16) & 0x7fff_ffff_ffff_0000L) |
886        (counter.getAndIncrement() & 0xffffL);
887  }
888
889}