HBASE-25824 IntegrationTestLoadCommonCrawl

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

@@ -0,0 +1,64 @@
+/*
+ * 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/warc/WARCFileReader.java

@@ -0,0 +1,165 @@
+/*
+ * 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.
+   */
+  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.
+   */
+  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.
+   */
+  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;
+    }
+  }
+}