HBASE-25824 IntegrationTestLoadCommonCrawl

This integration test loads successful resource retrieval records from
the Common Crawl (https://commoncrawl.org/) public dataset into an HBase
table and writes records that can be used to later verify the presence
and integrity of those records.

Run like:

  ./bin/hbase org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl \
    -Dfs.s3n.awsAccessKeyId=<AWS access key> \
    -Dfs.s3n.awsSecretAccessKey=<AWS secret key> \
    /path/to/test-CC-MAIN-2021-10-warc.paths.gz \
    /path/to/tmp/warc-loader-output

Access to the Common Crawl dataset in S3 is made available to anyone by
Amazon AWS, but Hadoop's S3N filesystem still requires valid access
credentials to initialize.

The input path can either specify a directory or a file. The file may
optionally be compressed with gzip. If a directory, the loader expects
the directory to contain one or more WARC files from the Common Crawl
dataset. If a file, the loader expects a list of Hadoop S3N URIs which
point to S3 locations for one or more WARC files from the Common Crawl
dataset, one URI per line. Lines should be terminated with the UNIX line
terminator.

Included in hbase-it/src/test/resources/CC-MAIN-2021-10-warc.paths.gz
is a list of all WARC files comprising the Q1 2021 crawl archive. There
are 64,000 WARC files in this data set, each containing ~1GB of gzipped
data. The WARC files contain several record types, such as metadata,
request, and response, but we only load the response record types. If
the HBase table schema does not specify compression (by default) there
is roughly a 10x expansion. Loading the full crawl archive results in a
table approximately 640 TB in size.

The hadoop-aws jar will be needed at runtime to instantiate the S3N
filesystem. Use the -files ToolRunner argument to add it.

You can also split the Loader and Verify stages:

Load with:

  ./bin/hbase 'org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl$Loader' \
    -files /path/to/hadoop-aws.jar \
    -Dfs.s3n.awsAccessKeyId=<AWS access key> \
    -Dfs.s3n.awsSecretAccessKey=<AWS secret key> \
    /path/to/test-CC-MAIN-2021-10-warc.paths.gz \
    /path/to/tmp/warc-loader-output

Verify with:

  ./bin/hbase 'org.apache.hadoop.hbase.test.IntegrationTestLoadCommonCrawl$Verify' \
    /path/to/tmp/warc-loader-output

Author: apurtell

hbase-it/src/test/java/org/apache/hadoop/hbase/test/IntegrationTestLoadCommonCrawl.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.test.util;
+
+// Cribbed from hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/CRC64.java
+
+public class CRC64 {
+  private static final long POLY = 0x9a6c9329ac4bc9b5L;
+  private static final int TABLE_LENGTH = 256;
+  private static final long[] TABLE = new long[TABLE_LENGTH];
+  static {
+  /* Initialize a table constructed from POLY  */
+    for (int n = 0; n < TABLE_LENGTH; ++n) {
+      long crc = n;
+      for (int i = 0; i < 8; ++i) {
+        if ((crc & 1) == 1) {
+          crc = (crc >>> 1) ^ POLY;
+        } else {
+          crc >>>= 1;
+        }
+      }
+      TABLE[n] = crc;
+    }
+  }
+
+  private long value = -1;
+
+  public void reset() {
+    value = -1;
+  }
+
+  public void update(byte[] input, int off, int len) {
+    for (int i = off; i < off+len; i++) {
+      value = TABLE[(input[i] ^ (int) value) & 0xFF] ^ (value >>> 8);
+    }
+  }
+
+  public void update(byte[] input) {
+    update(input, 0, input.length);
+  }
+
+  public long getValue() {
+    // Return the compliment of 'value' to complete the calculation
+    return ~value;
+  }
+
+}

hbase-it/src/test/java/org/apache/hadoop/hbase/test/util/CRC64.java

@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/*
+ * The MIT License (MIT)
+ * Copyright (c) 2014 Martin Kleppmann
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package org.apache.hadoop.hbase.test.util.warc;
+
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reads {@link WARCRecord}s from a WARC file, using Hadoop's filesystem APIs. (This means you
+ * can read from HDFS, S3 or any other filesystem supported by Hadoop). This implementation is
+ * not tied to the MapReduce APIs -- that link is provided by the mapred
+ * {@link com.martinkl.warc.mapred.WARCInputFormat} and the mapreduce
+ * {@link com.martinkl.warc.mapreduce.WARCInputFormat}.
+ */
+public class WARCFileReader {
+    private static final Logger logger = LoggerFactory.getLogger(WARCFileReader.class);
+
+    private final long fileSize;
+    private CountingInputStream byteStream = null;
+    private DataInputStream dataStream = null;
+    private long bytesRead = 0, recordsRead = 0;
+
+    /**
+     * Opens a file for reading. If the filename ends in `.gz`, it is automatically decompressed
+     * on the fly.
+     * @param conf The Hadoop configuration.
+     * @param filePath The Hadoop path to the file that should be read.
+     * @throws IOException
+     */
+    public WARCFileReader(Configuration conf, Path filePath) throws IOException {
+        FileSystem fs = filePath.getFileSystem(conf);
+        this.fileSize = fs.getFileStatus(filePath).getLen();
+        logger.info("Reading from " + filePath);
+
+        CompressionCodec codec = filePath.getName().endsWith(".gz") ?
+            WARCFileWriter.getGzipCodec(conf) : null;
+        byteStream = new CountingInputStream(new BufferedInputStream(fs.open(filePath)));
+        dataStream = new DataInputStream(codec == null ? byteStream : codec.createInputStream(byteStream));
+    }
+
+    /**
+     * Reads the next record from the file.
+     * @return The record that was read.
+     * @throws IOException
+     */
+    public WARCRecord read() throws IOException {
+        WARCRecord record = new WARCRecord(dataStream);
+        recordsRead++;
+        return record;
+    }
+
+    /**
+     * Closes the file. No more reading is possible after the file has been closed.
+     * @throws IOException
+     */
+    public void close() throws IOException {
+        if (dataStream != null) dataStream.close();
+        byteStream = null;
+        dataStream = null;
+    }
+
+    /**
+     * Returns the number of records that have been read since the file was opened.
+     */
+    public long getRecordsRead() {
+        return recordsRead;
+    }
+
+    /**
+     * Returns the number of bytes that have been read from file since it was opened.
+     * If the file is compressed, this refers to the compressed file size.
+     */
+    public long getBytesRead() {
+        return bytesRead;
+    }
+
+    /**
+     * Returns the proportion of the file that has been read, as a number between 0.0
+     * and 1.0.
+     */
+    public float getProgress() {
+        if (fileSize == 0) return 1.0f;
+        return (float) bytesRead / (float) fileSize;
+    }
+
+    private class CountingInputStream extends FilterInputStream {
+        public CountingInputStream(InputStream in) {
+            super(in);
+        }
+
+        @Override
+        public int read() throws IOException {
+            int result = in.read();
+            if (result != -1) bytesRead++;
+            return result;
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            int result = in.read(b, off, len);
+            if (result != -1) bytesRead += result;
+            return result;
+        }
+
+        @Override
+        public long skip(long n) throws IOException {
+            long result = in.skip(n);
+            bytesRead += result;
+            return result;
+        }
+    }
+}