lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-22 13:46:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'ait-qa' into adds-trough-dedup-performance-rework

Browse Source
This commit is contained in:
Adam Miller
2021-04-05 16:17:47 -07:00
committed by GitHub
parent 0b09340e84 21e84ad015
commit e353ccb097
31 changed files with 256 additions and 2448 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/main/java/org/archive/util/Base32.java
+18 -128
View File
@@ -18,142 +18,32 @@
*/
package org.archive.util;
import com.google.common.io.BaseEncoding;
/**
* Base32 - encodes and decodes RFC3548 Base32
* (see http://www.faqs.org/rfcs/rfc3548.html )
*
* Imported public-domain code of Bitzi.
*
* @author Robert Kaye
* @author Gordon Mohr
* @deprecated Use {@link com.google.common.io.BaseEncoding#base32()}
*/
@Deprecated
public class Base32 {
private static final String base32Chars =
"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567";
private static final int[] base32Lookup =
{ 0xFF,0xFF,0x1A,0x1B,0x1C,0x1D,0x1E,0x1F, // '0', '1', '2', '3', '4', '5', '6', '7'
0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF,0xFF, // '8', '9', ':', ';', '<', '=', '>', '?'
0xFF,0x00,0x01,0x02,0x03,0x04,0x05,0x06, // '@', 'A', 'B', 'C', 'D', 'E', 'F', 'G'
0x07,0x08,0x09,0x0A,0x0B,0x0C,0x0D,0x0E, // 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O'
0x0F,0x10,0x11,0x12,0x13,0x14,0x15,0x16, // 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W'
0x17,0x18,0x19,0xFF,0xFF,0xFF,0xFF,0xFF, // 'X', 'Y', 'Z', '[', '\', ']', '^', '_'
0xFF,0x00,0x01,0x02,0x03,0x04,0x05,0x06, // '`', 'a', 'b', 'c', 'd', 'e', 'f', 'g'
0x07,0x08,0x09,0x0A,0x0B,0x0C,0x0D,0x0E, // 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o'
0x0F,0x10,0x11,0x12,0x13,0x14,0x15,0x16, // 'p', 'q', 'r', 's', 't', 'u', 'v', 'w'
0x17,0x18,0x19,0xFF,0xFF,0xFF,0xFF,0xFF // 'x', 'y', 'z', '{', '|', '}', '~', 'DEL'
};
/**
* Encodes byte array to Base32 String.
*
* @param bytes Bytes to encode.
* @return Encoded byte array <code>bytes</code> as a String.
*
* @deprecated Use {@link com.google.common.io.BaseEncoding#base32()}
*/
@Deprecated
static public String encode(final byte[] bytes) {
int i = 0, index = 0, digit = 0;
int currByte, nextByte;
StringBuffer base32 = new StringBuffer((bytes.length + 7) * 8 / 5);
while (i < bytes.length) {
currByte = (bytes[i] >= 0) ? bytes[i] : (bytes[i] + 256); // unsign
/* Is the current digit going to span a byte boundary? */
if (index > 3) {
if ((i + 1) < bytes.length) {
nextByte =
(bytes[i + 1] >= 0) ? bytes[i + 1] : (bytes[i + 1] + 256);
} else {
nextByte = 0;
}
digit = currByte & (0xFF >> index);
index = (index + 5) % 8;
digit <<= index;
digit |= nextByte >> (8 - index);
i++;
} else {
digit = (currByte >> (8 - (index + 5))) & 0x1F;
index = (index + 5) % 8;
if (index == 0)
i++;
}
base32.append(base32Chars.charAt(digit));
}
return base32.toString();
return BaseEncoding.base32()
.omitPadding()
.lowerCase()
.encode(bytes)
.toUpperCase();
}
/**
* Decodes the given Base32 String to a raw byte array.
*
* @param base32
* @return Decoded <code>base32</code> String as a raw byte array.
* @deprecated Use {@link com.google.common.io.BaseEncoding#base32()}
*/
static public byte[] decode(final String base32) {
int i, index, lookup, offset, digit;
byte[] bytes = new byte[base32.length() * 5 / 8];
for (i = 0, index = 0, offset = 0; i < base32.length(); i++) {
lookup = base32.charAt(i) - '0';
/* Skip chars outside the lookup table */
if (lookup < 0 || lookup >= base32Lookup.length) {
continue;
}
digit = base32Lookup[lookup];
/* If this digit is not in the table, ignore it */
if (digit == 0xFF) {
continue;
}
if (index <= 3) {
index = (index + 5) % 8;
if (index == 0) {
bytes[offset] |= digit;
offset++;
if (offset >= bytes.length)
break;
} else {
bytes[offset] |= digit << (8 - index);
}
} else {
index = (index + 5) % 8;
bytes[offset] |= (digit >>> index);
offset++;
if (offset >= bytes.length) {
break;
}
bytes[offset] |= digit << (8 - index);
}
}
return bytes;
}
/** For testing, take a command-line argument in Base32, decode, print in hex,
* encode, print
*
* @param args
*/
static public void main(String[] args) {
if (args.length == 0) {
System.out.println("Supply a Base32-encoded argument.");
return;
}
System.out.println(" Original: " + args[0]);
byte[] decoded = Base32.decode(args[0]);
System.out.print(" Hex: ");
for (int i = 0; i < decoded.length; i++) {
int b = decoded[i];
if (b < 0) {
b += 256;
}
System.out.print((Integer.toHexString(b + 256)).substring(1));
}
System.out.println();
System.out.println("Reencoded: " + Base32.encode(decoded));
@Deprecated
static public byte[] decode(final String base32) {
return BaseEncoding.base32()
.omitPadding()
.lowerCase()
.decode(base32.toLowerCase());
}
}
commons/src/main/java/org/archive/util/BloomFilter64bit.java
+47 -213
View File
@@ -27,91 +27,30 @@
package org.archive.util;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.security.SecureRandom;
import java.util.Random;
/** A Bloom filter.
*
* ADAPTED/IMPROVED VERSION OF MG4J it.unimi.dsi.mg4j.util.BloomFilter
*
* <p>KEY CHANGES:
*
* <ul>
* <li>NUMBER_OF_WEIGHTS is 2083, to better avoid collisions between
* similar strings (common in the domain of URIs)</li>
*
* <li>Removed dependence on cern.colt MersenneTwister (replaced with
* SecureRandom) and QuickBitVector (replaced with local methods).</li>
*
* <li>Adapted to allow long bit indices</li>
*
* <li>Stores bitfield in an array of up to 2^22 arrays of 2^26 longs. Thus,
* bitfield may grow to 2^48 longs in size -- 2PiB, 2*54 bitfield indexes.
* (I expect this will outstrip available RAM for the next few years.)</li>
* </ul>
*
* <hr>
*
* <P>Instances of this class represent a set of character sequences (with
* false positives) using a Bloom filter. Because of the way Bloom filters work,
* you cannot remove elements.
*
* <P>Bloom filters have an expected error rate, depending on the number
* of hash functions used, on the filter size and on the number of elements in
* the filter. This implementation uses a variable optimal number of hash
* functions, depending on the expected number of elements. More precisely, a
* Bloom filter for <var>n</var> character sequences with <var>d</var> hash
* functions will use ln 2 <var>d</var><var>n</var> &#8776;
* 1.44 <var>d</var><var>n</var> bits; false positives will happen with
* probability 2<sup>-<var>d</var></sup>.
*
* <P>Hash functions are generated at creation time using universal hashing.
* Each hash function uses {@link #NUMBER_OF_WEIGHTS} random integers, which
* are cyclically multiplied by the character codes in a character sequence.
* The resulting integers are XOR-ed together.
*
* <P>This class exports access methods that are very similar to those of
* {@link java.util.Set}, but it does not implement that interface, as too
* many non-optional methods would be unimplementable (e.g., iterators).
*
* @author Sebastiano Vigna
* @author Gordon Mohr
*/
import com.google.common.annotations.VisibleForTesting;
import com.google.common.hash.Funnels;
import com.google.common.primitives.Ints;
public class BloomFilter64bit implements Serializable, BloomFilter {
private static final long serialVersionUID = 2L;
private static final long serialVersionUID = 3L;
/** The number of weights used to create hash functions. */
protected final static int NUMBER_OF_WEIGHTS = 2083; // CHANGED FROM 16
/** The number of bits in this filter. */
final protected long m;
/** if bitfield is an exact power of 2 in length, it is this power */
protected int power = -1;
/** The expected number of inserts; determines calculated size */
final protected long expectedInserts;
/** The number of hash functions used by this filter. */
final protected int d;
/** The underlying bit vector */
final protected long[][] bits;
/** The random integers used to generate the hash functions. */
final protected long[][] weight;
private final long expectedInserts;
/** The number of elements currently in the filter. It may be
* smaller than the actual number of additions of distinct character
* sequences because of false positives.
*/
protected int size;
private int size;
/** The natural logarithm of 2, used in the computation of the number of bits. */
protected final static double NATURAL_LOG_OF_2 = Math.log( 2 );
/** power-of-two to use as maximum size of bitfield subarrays */
protected final static int SUBARRAY_POWER_OF_TWO = 26; // 512MiB of longs
/** number of longs in one subarray */
protected final static int SUBARRAY_LENGTH_IN_LONGS = 1 << SUBARRAY_POWER_OF_TWO;
/** mask for lowest SUBARRAY_POWER_OF_TWO bits */
protected final static int SUBARRAY_MASK = SUBARRAY_LENGTH_IN_LONGS - 1; //0x0FFFFFFF
protected final static boolean DEBUG = false;
private final com.google.common.hash.BloomFilter<CharSequence> delegate;
private final long bitSize;
private final int numHashFunctions;
/** Creates a new Bloom filter with given number of hash functions and
* expected number of elements.
@@ -141,45 +80,18 @@ public class BloomFilter64bit implements Serializable, BloomFilter {
* @param roundUp if true, round bit size up to next-nearest-power-of-2
*/
public BloomFilter64bit(final long n, final int d, Random weightsGenerator, boolean roundUp ) {
delegate = com.google.common.hash.BloomFilter.create(Funnels.unencodedCharsFunnel(), Ints.saturatedCast(n), 0.0000003);
this.expectedInserts = n;
this.d = d;
long lenInLongs = (long)Math.ceil( ( (long)n * (long)d / NATURAL_LOG_OF_2 ) / 64L );
if ( lenInLongs > (1L<<48) ) {
throw new IllegalArgumentException(
"This filter would require " + lenInLongs + " longs, " +
"greater than this classes maximum of 2^48 longs (2PiB)." );
}
long lenInBits = lenInLongs * 64L;
if(roundUp) {
int pow = 0;
while((1L<<pow) < lenInBits) {
pow++;
}
this.power = pow;
this.m = 1L<<pow;
lenInLongs = m/64L;
} else {
this.m = lenInBits;
}
try {
Method bitSizeMethod = delegate.getClass().getDeclaredMethod("bitSize", new Class[] {});
bitSizeMethod.setAccessible(true);
bitSize = (long) bitSizeMethod.invoke(delegate, new Object[] {});
int arrayOfArraysLength = (int)((lenInLongs+SUBARRAY_LENGTH_IN_LONGS-1)/SUBARRAY_LENGTH_IN_LONGS);
bits = new long[ (int)(arrayOfArraysLength) ][];
// ensure last subarray is no longer than necessary
long lenInLongsRemaining = lenInLongs;
for(int i = 0; i < bits.length; i++) {
bits[i] = new long[(int)Math.min(lenInLongsRemaining,SUBARRAY_LENGTH_IN_LONGS)];
lenInLongsRemaining -= bits[i].length;
}
if ( DEBUG ) System.err.println( "Number of bits: " + m );
weight = new long[ d ][];
for( int i = 0; i < d; i++ ) {
weight[ i ] = new long[ NUMBER_OF_WEIGHTS ];
for( int j = 0; j < NUMBER_OF_WEIGHTS; j++ )
weight[ i ][ j ] = weightsGenerator.nextLong();
Field numHashFunctionField = delegate.getClass().getDeclaredField("numHashFunctions");
numHashFunctionField.setAccessible(true);
numHashFunctions = numHashFunctionField.getInt(delegate);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -193,36 +105,6 @@ public class BloomFilter64bit implements Serializable, BloomFilter {
return size;
}
/** Hashes the given sequence with the given hash function.
*
* @param s a character sequence.
* @param l the length of <code>s</code>.
* @param k a hash function index (smaller than {@link #d}).
* @return the position in the filter corresponding to <code>s</code> for the hash function <code>k</code>.
*/
protected long hash( final CharSequence s, final int l, final int k ) {
final long[] w = weight[ k ];
long h = 0;
int i = l;
while( i-- != 0 ) h ^= s.charAt( i ) * w[ i % NUMBER_OF_WEIGHTS ];
long retVal;
if(power>0) {
retVal = h >>> (64-power);
} else {
// ####----####----
retVal = ( h & 0x7FFFFFFFFFFFFFFFL ) % m;
}
return retVal;
}
public long[] bitIndexesFor(CharSequence s) {
long[] ret = new long[d];
for(int i = 0; i < d; i++) {
ret[i] = hash(s,s.length(),i);
}
return ret;
}
/** Checks whether the given character sequence is in this filter.
*
* <P>Note that this method may return true on a character sequence that is has
@@ -237,9 +119,7 @@ public class BloomFilter64bit implements Serializable, BloomFilter {
*/
public boolean contains( final CharSequence s ) {
int i = d, l = s.length();
while( i-- != 0 ) if ( ! getBit( hash( s, l, i ) ) ) return false;
return true;
return delegate.mightContain(s);
}
/** Adds a character sequence to the filter.
@@ -249,79 +129,19 @@ public class BloomFilter64bit implements Serializable, BloomFilter {
*/
public boolean add( final CharSequence s ) {
boolean result = false;
int i = d, l = s.length();
long h;
while( i-- != 0 ) {
h = hash( s, l, i );
if ( ! setGetBit( h ) ) {
result = true;
}
}
if ( result ) size++;
return result;
}
protected final static long ADDRESS_BITS_PER_UNIT = 6; // 64=2^6
protected final static long BIT_INDEX_MASK = (1<<6)-1; // = 63 = 2^BITS_PER_UNIT - 1;
/**
* Returns from the local bitvector the value of the bit with
* the specified index. The value is <tt>true</tt> if the bit
* with the index <tt>bitIndex</tt> is currently set; otherwise,
* returns <tt>false</tt>.
*
* (adapted from cern.colt.bitvector.QuickBitVector)
*
* @param bitIndex the bit index.
* @return the value of the bit with the specified index.
*/
public boolean getBit(long bitIndex) {
long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT;
int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO);
int subarrayIndex = (int) (longIndex & SUBARRAY_MASK);
return ((bits[arrayIndex][subarrayIndex] & (1L << (bitIndex & BIT_INDEX_MASK))) != 0);
boolean added = delegate.put(s);
if (added) {
size++;
}
return added;
}
/**
* Changes the bit with index <tt>bitIndex</tt> in local bitvector.
*
* (adapted from cern.colt.bitvector.QuickBitVector)
*
* @param bitIndex the index of the bit to be set.
/* (non-Javadoc)
* @see org.archive.util.BloomFilter#getSizeBytes()
*/
protected void setBit( long bitIndex) {
long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT;
int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO);
int subarrayIndex = (int) (longIndex & SUBARRAY_MASK);
bits[arrayIndex][subarrayIndex] |= (1L << (bitIndex & BIT_INDEX_MASK));
public long getSizeBytes() {
return bitSize / 8;
}
/**
* Sets the bit with index <tt>bitIndex</tt> in local bitvector --
* returning the old value.
*
* (adapted from cern.colt.bitvector.QuickBitVector)
*
* @param bitIndex the index of the bit to be set.
*/
protected boolean setGetBit( long bitIndex) {
long longIndex = bitIndex >>> ADDRESS_BITS_PER_UNIT;
int arrayIndex = (int) (longIndex >>> SUBARRAY_POWER_OF_TWO);
int subarrayIndex = (int) (longIndex & SUBARRAY_MASK);
long mask = 1L << (bitIndex & BIT_INDEX_MASK);
boolean ret = (bits[arrayIndex][subarrayIndex] & mask)!=0;
bits[arrayIndex][subarrayIndex] |= mask;
return ret;
}
/* (non-Javadoc)
* @see org.archive.util.BloomFilter#getSizeBytes()
*/
public long getSizeBytes() {
// account for ragged-sized last array
return 8*(((bits.length-1)*bits[0].length)+bits[bits.length-1].length);
}
@Override
public long getExpectedInserts() {
@@ -330,6 +150,20 @@ public class BloomFilter64bit implements Serializable, BloomFilter {
@Override
public long getHashCount() {
return d;
return numHashFunctions;
}
@VisibleForTesting
public boolean getBit(long bitIndex) {
try {
Field bitsField = delegate.getClass().getDeclaredField("bits");
bitsField.setAccessible(true);
Object bitarray = bitsField.get(delegate);
Method getBitMethod = bitarray.getClass().getDeclaredMethod("get", long.class);
getBitMethod.setAccessible(true);
return (boolean) getBitMethod.invoke(bitarray, bitIndex);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
contrib/pom.xml
-22
View File
@@ -13,28 +13,6 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>0.98.6-cdh5.3.5</version>
<exclusions>
<exclusion>
<artifactId>jets3t</artifactId>
<groupId>net.java.dev.jets3t</groupId>
</exclusion>
<exclusion>
<artifactId>junit</artifactId>
<groupId>junit</groupId>
</exclusion>
<!-- tools.jar is not available in JDK 11 so exclude it
hbase-client accidentally leaked it as a transitive dependency
https://issues.apache.org/jira/browse/HBASE-13963 -->
<exclusion>
<groupId>jdk.tools</groupId>
<artifactId>jdk.tools</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.archive.heritrix</groupId>
<artifactId>heritrix-engine</artifactId>
contrib/src/main/java/org/archive/modules/extractor/ExtractorYoutubeDL.java
+70 -17
View File
@@ -33,6 +33,7 @@ import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.nio.channels.Channels;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
@@ -44,6 +45,7 @@ import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.URIException;
import org.archive.crawler.frontier.AMQPUrlReceiver;
import org.archive.crawler.reporting.CrawlerLoggerModule;
import org.archive.format.warc.WARCConstants.WARCRecordType;
import org.archive.io.warc.WARCRecordInfo;
@@ -112,22 +114,60 @@ public class ExtractorYoutubeDL extends Extractor
protected static final int MAX_VIDEOS_PER_PAGE = 1000;
// for shouldExtract
protected HashMap<String, Boolean> seedsYDLd = new HashMap<String, Boolean>();
protected transient Logger ydlLogger = null;
// unnamed toethread-local temporary file
protected transient ThreadLocal<RandomAccessFile> tempfile = new ThreadLocal<RandomAccessFile>() {
protected RandomAccessFile initialValue() {
File t;
try {
t = File.createTempFile("ydl", ".json");
RandomAccessFile f = new RandomAccessFile(t, "rw");
t.delete();
return f;
} catch (IOException e) {
throw new RuntimeException(e);
}
return null;
}
};
protected void closeLocalTempFile() {
RandomAccessFile localTemp = tempfile.get();
if(localTemp == null || !isOpen(localTemp))
return; // avoid making a new temp file just to close it immediately
try {
getLocalTempFile().close();
tempfile.set(null);
}
catch (Exception e) {
logger.log(Level.WARNING, "problem closing ydl temp file " + e);
}
}
protected RandomAccessFile getLocalTempFile() {
RandomAccessFile localTemp = tempfile.get();
if(localTemp == null || !isOpen(localTemp)) {
localTemp = openNewTempFile();
tempfile.set(localTemp);
}
logger.info("Getting youtube-dl temp file ");
return localTemp;
}
protected boolean isOpen(RandomAccessFile f) {
try {
f.length();
return true;
}
catch (IOException e) {
logger.info("youtube-dl temp file is not open");
return false ;
}
}
protected RandomAccessFile openNewTempFile() {
logger.info("Opening New youtube-dl temp file ");
File t;
try {
t = File.createTempFile("ydl", ".json");
RandomAccessFile f = new RandomAccessFile(t, "rw");
t.delete();
return f;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
protected CrawlerLoggerModule crawlerLoggerModule;
public CrawlerLoggerModule getCrawlerLoggerModule() {
@@ -419,7 +459,8 @@ public class ExtractorYoutubeDL extends Extractor
* https://github.com/ytdl-org/youtube-dl/blob/master/README.md#format-selection
*/
ProcessBuilder pb = new ProcessBuilder("youtube-dl", "--ignore-config",
"--simulate", "--dump-single-json", "--format=best",
"--simulate", "--dump-single-json", "--format=best[height <=? 576]",
"--no-cache-dir", "--no-playlist",
"--playlist-end=" + MAX_VIDEOS_PER_PAGE, uri.toString());
logger.info("running: " + String.join(" ", pb.command()));
@@ -446,7 +487,7 @@ public class ExtractorYoutubeDL extends Extractor
}
});
YoutubeDLResults results = new YoutubeDLResults(tempfile.get());
YoutubeDLResults results = new YoutubeDLResults(getLocalTempFile());
try {
try {
@@ -507,6 +548,11 @@ public class ExtractorYoutubeDL extends Extractor
return false;
}
// skip crawl uris received from umbra
if (uri.getAnnotations().contains(AMQPUrlReceiver.A_RECEIVED_FROM_AMQP)) {
return false;
}
String mime = uri.getContentType().toLowerCase();
if (mime.startsWith("text/html")
|| mime.startsWith("application/xhtml")
@@ -524,7 +570,14 @@ public class ExtractorYoutubeDL extends Extractor
// should build record for containing page, which has an
// annotation like "youtube-dl:3" (no slash)
String annotation = findYdlAnnotation(uri);
return annotation != null && !annotation.contains("/");
boolean shouldBuild = (annotation != null && !annotation.contains("/"));
// If we processed this uri, then we have an open temp file that won't get closed
// for us by the warc writer
if(!shouldBuild)
closeLocalTempFile();
return shouldBuild;
}
@Override
@@ -545,10 +598,10 @@ public class ExtractorYoutubeDL extends Extractor
recordInfo.setMimetype("application/vnd.youtube-dl_formats+json;charset=utf-8");
recordInfo.setEnforceLength(true);
tempfile.get().seek(0);
InputStream inputStream = Channels.newInputStream(tempfile.get().getChannel());
getLocalTempFile().seek(0);
InputStream inputStream = Channels.newInputStream(getLocalTempFile().getChannel());
recordInfo.setContentStream(inputStream);
recordInfo.setContentLength(tempfile.get().length());
recordInfo.setContentLength(getLocalTempFile().length());
logger.info("built record timestamp=" + timestamp + " url=" + recordInfo.getUrl());
@@ -574,7 +627,7 @@ public class ExtractorYoutubeDL extends Extractor
ExtractorYoutubeDL e = new ExtractorYoutubeDL();
FileInputStream in = new FileInputStream("/tmp/ydl-single-video.json");
YoutubeDLResults results = new YoutubeDLResults(e.tempfile.get());
YoutubeDLResults results = new YoutubeDLResults(e.getLocalTempFile());
e.streamYdlOutput(in, results);
System.out.println("video urls: " + results.videoUrls);
System.out.println("page urls: " + results.pageUrls);
@@ -590,7 +643,7 @@ public class ExtractorYoutubeDL extends Extractor
}
in = new FileInputStream("/tmp/ydl-uncgreensboro-limited.json");
results = new YoutubeDLResults(e.tempfile.get());
results = new YoutubeDLResults(e.getLocalTempFile());
e.streamYdlOutput(in, results);
System.out.println("video urls: " + results.videoUrls);
System.out.println("page urls: " + results.pageUrls);
contrib/src/main/java/org/archive/modules/postprocessor/KafkaCrawlLogFeed.java
-255
View File
@@ -1,255 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.postprocessor;
import java.io.UnsupportedEncodingException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.logging.Logger;
import org.apache.commons.collections.Closure;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.archive.crawler.framework.Frontier;
import org.archive.crawler.frontier.AbstractFrontier;
import org.archive.crawler.frontier.BdbFrontier;
import org.archive.crawler.io.UriProcessingFormatter;
import org.archive.modules.CrawlURI;
import org.archive.modules.Processor;
import org.archive.modules.net.ServerCache;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.Lifecycle;
/**
* For Kafka 0.8.x. Sends messages in asynchronous mode (producer.type=async)
* and does not wait for acknowledgment from kafka (request.required.acks=0).
* Sends messages with no key. These things could be configurable if needed.
*
* @see UriProcessingFormatter
* @author nlevitt
*/
public class KafkaCrawlLogFeed extends Processor implements Lifecycle {
protected static final Logger logger = Logger.getLogger(KafkaCrawlLogFeed.class.getName());
protected Frontier frontier;
public Frontier getFrontier() {
return this.frontier;
}
/** Autowired frontier, needed to determine when a url is finished. */
@Autowired
public void setFrontier(Frontier frontier) {
this.frontier = frontier;
}
protected ServerCache serverCache;
public ServerCache getServerCache() {
return this.serverCache;
}
@Autowired
public void setServerCache(ServerCache serverCache) {
this.serverCache = serverCache;
}
protected Map<String,String> extraFields;
public Map<String, String> getExtraFields() {
return extraFields;
}
public void setExtraFields(Map<String, String> extraFields) {
this.extraFields = extraFields;
}
protected boolean dumpPendingAtClose = false;
public boolean getDumpPendingAtClose() {
return dumpPendingAtClose;
}
/**
* If true, publish all pending urls (i.e. queued urls still in the
* frontier) when crawl job is stopping. They are recognizable by the status
* field which has the value 0.
*
* @see BdbFrontier#setDumpPendingAtClose(boolean)
*/
public void setDumpPendingAtClose(boolean dumpPendingAtClose) {
this.dumpPendingAtClose = dumpPendingAtClose;
}
protected String brokerList = "localhost:9092";
/** Kafka broker list (kafka property "metadata.broker.list"). */
public void setBrokerList(String brokerList) {
this.brokerList = brokerList;
}
public String getBrokerList() {
return brokerList;
}
protected String topic = "heritrix-crawl-log";
public void setTopic(String topic) {
this.topic = topic;
}
public String getTopic() {
return topic;
}
protected byte[] buildMessage(CrawlURI curi) {
JSONObject jo = CrawlLogJsonBuilder.buildJson(curi, getExtraFields(), getServerCache());
try {
return jo.toString().getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
}
@Override
protected boolean shouldProcess(CrawlURI curi) {
if (frontier instanceof AbstractFrontier) {
return !((AbstractFrontier) frontier).needsReenqueuing(curi);
} else {
return false;
}
}
private transient long pendingDumpedCount = 0l;
@Override
public synchronized void stop() {
if (!isRunning) {
return;
}
if (dumpPendingAtClose) {
if (frontier instanceof BdbFrontier) {
Closure closure = new Closure() {
public void execute(Object curi) {
try {
innerProcess((CrawlURI) curi);
pendingDumpedCount++;
} catch (InterruptedException e) {
}
}
};
logger.info("dumping " + frontier.queuedUriCount() + " queued urls to kafka feed");
((BdbFrontier) frontier).forAllPendingDo(closure);
logger.info("dumped " + pendingDumpedCount + " queued urls to kafka feed");
} else {
logger.warning("frontier is not a BdbFrontier, cannot dumpPendingAtClose");
}
}
String rateStr = String.format("%1.1f", 0.01 * stats.errors / stats.total);
logger.info("final error count: " + stats.errors + "/" + stats.total + " (" + rateStr + "%)");
if (kafkaProducer != null) {
kafkaProducer.close();
kafkaProducer = null;
}
if (kafkaProducerThreads != null) {
kafkaProducerThreads.destroy();
kafkaProducerThreads = null;
}
super.stop();
}
private transient ThreadGroup kafkaProducerThreads;
transient protected KafkaProducer<String, byte[]> kafkaProducer;
protected KafkaProducer<String, byte[]> kafkaProducer() {
if (kafkaProducer == null) {
synchronized (this) {
if (kafkaProducer == null) {
final Properties props = new Properties();
props.put("bootstrap.servers", getBrokerList());
props.put("acks", "1");
props.put("producer.type", "async");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", ByteArraySerializer.class.getName());
/*
* XXX This mess here exists so that the kafka producer
* thread is in a thread group that is not the ToePool,
* so that it doesn't get interrupted at the end of the
* crawl in ToePool.cleanup().
*/
kafkaProducerThreads = new ThreadGroup(Thread.currentThread().getThreadGroup().getParent(), "KafkaProducerThreads");
ThreadFactory threadFactory = new ThreadFactory() {
public Thread newThread(Runnable r) {
return new Thread(kafkaProducerThreads, r);
}
};
Callable<KafkaProducer<String,byte[]>> task = new Callable<KafkaProducer<String,byte[]>>() {
public KafkaProducer<String, byte[]> call() throws InterruptedException {
return new KafkaProducer<String,byte[]>(props);
}
};
ExecutorService executorService = Executors.newFixedThreadPool(1, threadFactory);
Future<KafkaProducer<String, byte[]>> future = executorService.submit(task);
try {
kafkaProducer = future.get();
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
throw new RuntimeException(e);
} finally {
executorService.shutdown();
}
}
}
}
return kafkaProducer;
}
protected final class StatsCallback implements Callback {
public long errors = 0l;
public long total = 0l;
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
total++;
if (exception != null) {
errors++;
}
if (total % 10000 == 0) {
String rateStr = String.format("%1.1f", 0.01 * errors / total);
logger.info("error count so far: " + errors + "/" + total + " (" + rateStr + "%)");
}
}
}
protected StatsCallback stats = new StatsCallback();
@Override
protected void innerProcess(CrawlURI curi) throws InterruptedException {
byte[] message = buildMessage(curi);
ProducerRecord<String, byte[]> producerRecord = new ProducerRecord<String,byte[]>(getTopic(), message);
kafkaProducer().send(producerRecord, stats);
}
}
contrib/src/main/java/org/archive/modules/postprocessor/TroughCrawlLogFeed.java
+23 -22
View File
@@ -93,7 +93,7 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle {
protected static final Logger logger = Logger.getLogger(TroughCrawlLogFeed.class.getName());
protected static final int BATCH_MAX_TIME_MS = 60 * 1000;
protected static final int BATCH_MAX_TIME_MS = 20 * 1000;
protected static final int BATCH_MAX_SIZE = 400;
protected AtomicInteger crawledBatchSize = new AtomicInteger(0);
protected AtomicInteger uncrawledBatchSize = new AtomicInteger(0);
@@ -333,28 +333,29 @@ public class TroughCrawlLogFeed extends Processor implements Lifecycle {
}
}
if(batch != null && batch.size() > 0) {
StringBuffer sqlTmpl = new StringBuffer();
sqlTmpl.append(
"insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)"
+ " values (%s, %s, %s, %s, %s, %s, %s)");
StringBuffer sqlTmpl = new StringBuffer();
sqlTmpl.append(
"insert into uncrawled_url (timestamp, url, hop_path, status_code, via, seed, host)"
+ " values (%s, %s, %s, %s, %s, %s, %s)");
for (int i = 1; i < batch.size(); i++) {
sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)");
}
Object[] flattenedValues = new Object[7 * batch.size()];
for (int i = 0; i < batch.size(); i++) {
System.arraycopy(batch.get(i),0,flattenedValues,7 * i, 7);
}
try {
troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues);
} catch (Exception e) {
logger.log(Level.WARNING, "problem posting batch of " + batch.size() + " uncrawled urls to trough segment " + getSegmentId(), e);
}
uncrawledBatchLastTime = System.currentTimeMillis();
uncrawledBatch.clear();
for (int i = 1; i < batch.size(); i++) {
sqlTmpl.append(", (%s, %s, %s, %s, %s, %s, %s)");
}
Object[] flattenedValues = new Object[7 * batch.size()];
for (int i = 0; i < batch.size(); i++) {
System.arraycopy(batch.get(i),0,flattenedValues,7 * i, 7);
}
try {
troughClient().write(getSegmentId(), sqlTmpl.toString(), flattenedValues);
} catch (Exception e) {
logger.log(Level.WARNING, "problem posting batch of " + batch.size() + " uncrawled urls to trough segment " + getSegmentId(), e);
}
uncrawledBatchLastTime = System.currentTimeMillis();
uncrawledBatch.clear();
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBase.java
-126
View File
@@ -1,126 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.Map;
import java.util.Map.Entry;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.springframework.context.Lifecycle;
/**
* Represents a deployment of HBase. (An instance, a database, an HBase...)
*
* @author nlevitt
*/
public class HBase implements Lifecycle {
private static final Logger logger =
Logger.getLogger(HBase.class.getName());
protected Configuration conf = null;
private Map<String,String> properties;
public Map<String,String> getProperties() {
return properties;
}
public void setProperties(Map<String,String> properties) {
this.properties = properties;
if (conf == null) {
conf = HBaseConfiguration.create();
}
for (Entry<String, String> entry: getProperties().entrySet()) {
conf.set(entry.getKey(), entry.getValue());
}
}
public synchronized Configuration configuration() {
if (conf == null) {
conf = HBaseConfiguration.create();
}
return conf;
}
// convenience setters
public void setZookeeperQuorum(String value) {
configuration().set("hbase.zookeeper.quorum", value);
}
public void setZookeeperClientPort(int port) {
configuration().setInt("hbase.zookeeper.property.clientPort", port);
}
protected transient HBaseAdmin admin;
public synchronized HBaseAdmin admin() throws IOException {
if (admin == null) {
admin = new HBaseAdmin(configuration());
}
return admin;
}
@Override
public synchronized void stop() {
isRunning = false;
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
}
admin = null;
}
if (conf != null) {
// HConnectionManager.deleteConnection(conf); // XXX?
conf = null;
}
}
protected transient boolean isRunning = false;
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void start() {
isRunning = true;
}
public synchronized void reset() {
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
}
admin = null;
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseContentDigestHistory.java
-303
View File
@@ -1,303 +0,0 @@
package org.archive.modules.recrawl.hbase;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_COUNT;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_DATE;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILENAME;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILE_OFFSET;
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_RECORD_ID;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.recrawl.AbstractContentDigestHistory;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.context.Lifecycle;
import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
/**
* HBase content digest history store. Must be a toplevel bean in
* crawler-beans.cxml in order to receive {@link Lifecycle} events.
*
* @see AbstractContentDigestHistory
* @author nlevitt
*/
public class HBaseContentDigestHistory extends AbstractContentDigestHistory implements Lifecycle {
private static final Logger logger =
Logger.getLogger(HBaseContentDigestHistory.class.getName());
protected static final byte[] COLUMN_FAMILY = Bytes.toBytes("f");
protected static final byte[] COLUMN = Bytes.toBytes("c");
protected static final BiMap<String,String> JSON_KEYS_MAP = HashBiMap.create();
static {
JSON_KEYS_MAP.put(A_CONTENT_DIGEST_COUNT, "c");
JSON_KEYS_MAP.put(A_ORIGINAL_URL, "u");
JSON_KEYS_MAP.put(A_WARC_RECORD_ID, "i");
JSON_KEYS_MAP.put(A_WARC_FILENAME, "f");
JSON_KEYS_MAP.put(A_WARC_FILE_OFFSET, "o");
JSON_KEYS_MAP.put(A_ORIGINAL_DATE, "d");
}
protected HBaseTable table;
public void setTable(HBaseTable table) {
this.table = table;
}
protected boolean addColumnFamily = false;
public boolean getAddColumnFamily() {
return addColumnFamily;
}
/**
* Add the expected column family
* {@link #COLUMN_FAMILY} to the HBase table if the
* table doesn't already have it.
*/
public void setAddColumnFamily(boolean addColumnFamily) {
this.addColumnFamily = addColumnFamily;
}
protected int retryIntervalMs = 10*1000;
public int getRetryIntervalMs() {
return retryIntervalMs;
}
public void setRetryIntervalMs(int retryIntervalMs) {
this.retryIntervalMs = retryIntervalMs;
}
protected int maxTries = 1;
public int getMaxTries() {
return maxTries;
}
public void setMaxTries(int maxTries) {
this.maxTries = maxTries;
}
protected String keySuffix = null;
public String getKeySuffix() {
return keySuffix;
}
/**
* If not null, keySuffix is appended to the lookup key when loading and
* storing digest history. Thus the key looks like {digest}{keySuffix}, e.g.
* "sha1:22SFHXERHNFOEY6WK7YOUN4PFIPZSB4D-1193". The purpose is to support
* multiple namespaces in a single hbase table, to avoid proliferation of
* small tables. The reason we use a suffix instead of a prefix is to leave
* open the possibility of deduplication across these different namespaces
* at some point in the future.
*
* @param keySuffix
*/
public void setKeySuffix(String keySuffix) {
this.keySuffix = keySuffix;
}
@Override
protected String persistKeyFor(CrawlURI curi) {
if (keySuffix != null) {
return super.persistKeyFor(curi) + keySuffix;
} else {
return super.persistKeyFor(curi);
}
}
protected synchronized void addColumnFamily() {
try {
HTableDescriptor oldDesc = table.getHtableDescriptor();
if (oldDesc.getFamily(COLUMN_FAMILY) == null) {
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
newDesc.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
HBaseAdmin hbaseAdmin = table.getHbase().admin();
hbaseAdmin.disableTable(table.getName());
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
hbaseAdmin.enableTable(table.getName());
}
} catch (IOException e) {
logger.warning("problem adding column family: " + e);
}
}
private boolean isRunning;
@Override
public void start() {
// add column family here to avoid disabling table while another
// ToeThread is trying to use it
if (getAddColumnFamily()) {
addColumnFamily();
}
this.isRunning = true;
}
@Override
public void stop() {
this.isRunning = false;
}
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void load(CrawlURI curi) {
// make this call in all cases so that the value is initialized and
// WARCWriterProcessor knows it should put the info in there
HashMap<String, Object> contentDigestHistory = curi.getContentDigestHistory();
byte[] key = Bytes.toBytes(persistKeyFor(curi));
Result hbaseResult = tryHbaseGet(curi, new Get(key));
if (hbaseResult != null) {
Map<String, Object> loadedHistory = parseHbaseResult(curi, hbaseResult);
if (loadedHistory != null) {
if (logger.isLoggable(Level.FINER)) {
logger.finer("loaded history by digest " + persistKeyFor(curi)
+ " for uri " + curi + " - " + loadedHistory);
}
contentDigestHistory.putAll(loadedHistory);
}
}
}
protected Result tryHbaseGet(CrawlURI curi, Get hbaseGet) {
try {
return table.get(hbaseGet);
} catch (IOException e) {
logger.warning("problem retrieving persist data from hbase, proceeding without, for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
return null;
}
}
protected Map<String, Object> parseHbaseResult(CrawlURI curi, Result hbaseResult) {
HashMap<String, Object> loadedHistory = null;
// no data for uri is indicated by empty Result
if (!hbaseResult.isEmpty()) {
byte[] jsonBytes = hbaseResult.getValue(COLUMN_FAMILY, COLUMN);
if (jsonBytes != null) {
JSONObject json = null;
try {
json = new JSONObject(Bytes.toString(jsonBytes));
loadedHistory = new HashMap<String,Object>();
@SuppressWarnings("unchecked")
Iterator<String> keyIter = json.keys();
while (keyIter.hasNext()) {
String jsonKey = keyIter.next();
Object jsonValue = json.get(jsonKey);
String historyMapKey = JSON_KEYS_MAP.inverse().get(jsonKey);
if (historyMapKey == null) {
logger.warning("unknown key \"" + jsonKey + "\" found in hbase json for digest " + persistKeyFor(curi));
historyMapKey = jsonKey;
}
loadedHistory.put(historyMapKey, jsonValue);
}
} catch (JSONException e) {
logger.warning("problem parsing json for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
}
} else {
// shouldn't happen? result.isEmpty() is normal case
logger.fine("[jsonBytes==null] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
}
} else {
logger.finest("[result.isEmpty()] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
}
return loadedHistory;
}
@Override
public void store(CrawlURI curi) {
if (!curi.hasContentDigestHistory()
|| curi.getContentDigestHistory().isEmpty()) {
return;
}
if (logger.isLoggable(Level.FINER)) {
logger.finer("storing history by digest " + persistKeyFor(curi)
+ " for uri " + curi + " - "
+ curi.getContentDigestHistory());
}
Put hbasePut = createHbasePut(curi);
tryHbasePut(curi, hbasePut);
}
protected void tryHbasePut(CrawlURI curi, Put p) {
int tryCount = 0;
do {
tryCount++;
try {
table.put(p);
return;
} catch (RetriesExhaustedWithDetailsException e) {
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
addColumnFamily();
tryCount--;
} else {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
}
} catch (IOException e) {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
} catch (NullPointerException e) {
// HTable.put() throws NullPointerException while connection is lost.
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + curi + " - " + e);
}
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
try {
Thread.sleep(getRetryIntervalMs());
} catch (InterruptedException ex) {
logger.warning("thread interrupted. aborting retry for " + curi);
return;
}
}
} while (tryCount < getMaxTries() && isRunning());
if (isRunning()) {
logger.warning("giving up after " + tryCount + " tries on put for " + curi);
}
}
protected Put createHbasePut(CrawlURI curi) {
byte[] key = Bytes.toBytes(persistKeyFor(curi));
Put hbasePut = new Put(key);
try {
JSONObject json = new JSONObject();
for (Entry<String, Object> entry: curi.getContentDigestHistory().entrySet()) {
String jsonKey = JSON_KEYS_MAP.get(entry.getKey());
if (jsonKey == null) {
logger.warning("unknown key \"" + entry.getKey() + "\" found in content digest history map for " + curi);
jsonKey = entry.getKey();
}
json.put(jsonKey, entry.getValue());
}
hbasePut.add(COLUMN_FAMILY, COLUMN, Bytes.toBytes(json.toString()));
} catch (JSONException e) {
// should not happen - all values are either primitive or String.
logger.log(Level.SEVERE, "problem creating json object for digest " + persistKeyFor(curi) + " uri " + curi, e);
}
return hbasePut;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistLoadProcessor.java
-88
View File
@@ -1,88 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.CrawlURI;
import org.archive.modules.ProcessResult;
import org.archive.modules.Processor;
import org.archive.modules.recrawl.FetchHistoryProcessor;
/**
* A {@link Processor} for retrieving recrawl info from HBase table.
* See {@link HBasePersistProcessor} for table schema.
* As with other fetch history processors, this needs to be combined with {@link FetchHistoryProcessor}
* (set up after FetchHTTP, before WarcWriter) to work.
* @see HBasePersistStoreProcessor
* @author kenji
*/
public class HBasePersistLoadProcessor extends HBasePersistProcessor {
private static final Logger logger =
Logger.getLogger(HBasePersistLoadProcessor.class.getName());
@Override
protected ProcessResult innerProcessResult(CrawlURI uri) throws InterruptedException {
byte[] key = rowKeyForURI(uri);
Get g = new Get(key);
try {
Result r = table.get(g);
// no data for uri is indicated by empty Result
if (r.isEmpty()) {
if (logger.isLoggable(Level.FINE)) {
logger.fine(uri + ": <no crawlinfo>");
}
return ProcessResult.PROCEED;
}
schema.load(r, uri);
if (uri.getFetchStatus() < 0) {
return ProcessResult.FINISH;
}
} catch (IOException e) {
logger.warning("problem retrieving persist data from hbase, proceeding without, for " + uri + " - " + e);
} catch (Exception ex) {
// get() throws RuntimeException upon ZooKeeper connection failures.
// no crawl history load failure should make fetch of URL fail.
logger.log(Level.WARNING, "Get failed for " + uri + ": ", ex);
}
return ProcessResult.PROCEED;
}
/**
* unused.
*/
@Override
protected void innerProcess(CrawlURI uri) throws InterruptedException {
}
@Override
protected boolean shouldProcess(CrawlURI uri) {
// TODO: we want deduplicate robots.txt, too.
//if (uri.isPrerequisite()) return false;
String scheme = uri.getUURI().getScheme();
if (!(scheme.equals("http") || scheme.equals("https") || scheme.equals("ftp"))) {
return false;
}
return true;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistProcessor.java
-32
View File
@@ -1,32 +0,0 @@
package org.archive.modules.recrawl.hbase;
import org.archive.modules.CrawlURI;
import org.archive.modules.recrawl.AbstractPersistProcessor;
import org.springframework.beans.factory.annotation.Required;
/**
* A base class for processors for keeping de-duplication data in HBase.
* Table schema is defined by {@link RecrawlDataSchema} implementation.
* @author kenji
*/
public abstract class HBasePersistProcessor extends AbstractPersistProcessor {
protected HBaseTableBean table;
@Required
public void setTable(HBaseTableBean table) {
this.table = table;
}
protected RecrawlDataSchema schema;
public RecrawlDataSchema getSchema() {
return schema;
}
@Required
public void setSchema(RecrawlDataSchema schema) {
this.schema = schema;
}
protected byte[] rowKeyForURI(CrawlURI curi) {
return schema.rowKeyForURI(curi);
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBasePersistStoreProcessor.java
-133
View File
@@ -1,133 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
/**
* @author kenji
*/
public class HBasePersistStoreProcessor extends HBasePersistProcessor implements FetchStatusCodes, RecrawlAttributeConstants {
private static final Logger logger = Logger.getLogger(HBasePersistStoreProcessor.class.getName());
protected boolean addColumnFamily = false;
public boolean getAddColumnFamily() {
return addColumnFamily;
}
/**
* Add the expected column family
* {@link HBaseContentDigestHistory#COLUMN_FAMILY} to the HBase table if the
* table doesn't already have it.
*/
public void setAddColumnFamily(boolean addColumnFamily) {
this.addColumnFamily = addColumnFamily;
}
protected int retryIntervalMs = 10*1000;
public int getRetryIntervalMs() {
return retryIntervalMs;
}
public void setRetryIntervalMs(int retryIntervalMs) {
this.retryIntervalMs = retryIntervalMs;
}
protected int maxTries = 1;
public int getMaxTries() {
return maxTries;
}
public void setMaxTries(int maxTries) {
this.maxTries = maxTries;
}
protected synchronized void addColumnFamily() {
try {
HTableDescriptor oldDesc = table.getHtableDescriptor();
byte[] columnFamily = Bytes.toBytes(schema.getColumnFamily());
if (oldDesc.getFamily(columnFamily) == null) {
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
newDesc.addFamily(new HColumnDescriptor(columnFamily));
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
HBaseAdmin hbaseAdmin = table.getHbase().admin();
hbaseAdmin.disableTable(table.getName());
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
hbaseAdmin.enableTable(table.getName());
}
} catch (IOException e) {
logger.warning("problem adding column family: " + e);
}
}
@Override
protected void innerProcess(CrawlURI uri) {
Put p = schema.createPut(uri);
int tryCount = 0;
do {
tryCount++;
try {
table.put(p);
return;
} catch (RetriesExhaustedWithDetailsException e) {
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
addColumnFamily();
tryCount--;
} else {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
}
} catch (IOException e) {
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
} catch (NullPointerException e) {
// HTable.put() throws NullPointerException while connection is lost.
logger.warning("put failed " + "(try " + tryCount + " of "
+ getMaxTries() + ")" + " for " + uri + " - " + e);
}
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
try {
Thread.sleep(getRetryIntervalMs());
} catch (InterruptedException ex) {
logger.warning("thread interrupted. aborting retry for " + uri);
return;
}
}
} while (tryCount < getMaxTries() && isRunning());
if (isRunning()) {
logger.warning("giving up after " + tryCount + " tries on put for " + uri);
}
}
@Override
protected boolean shouldProcess(CrawlURI curi) {
return super.shouldStore(curi);
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTable.java
-159
View File
@@ -1,159 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
/**
* @author kenji
* @author nlevitt
*/
public class HBaseTable extends HBaseTableBean {
static final Logger logger =
Logger.getLogger(HBaseTable.class.getName());
protected boolean create = false;
protected HConnection hconn = null;
protected ThreadLocal<HTableInterface> htable = new ThreadLocal<HTableInterface>();
public boolean getCreate() {
return create;
}
/** Create the named table if it doesn't exist. */
public void setCreate(boolean create) {
this.create = create;
}
public HBaseTable() {
}
protected synchronized HConnection hconnection() throws IOException {
if (hconn == null) {
hconn = HConnectionManager.createConnection(hbase.configuration());
}
return hconn;
}
protected HTableInterface htable() throws IOException {
if (htable.get() == null) {
htable.set(hconnection().getTable(htableName));
}
return htable.get();
}
@Override
public void put(Put p) throws IOException {
try {
htable().put(p);
} catch (IOException e) {
reset();
throw e;
}
}
@Override
public Result get(Get g) throws IOException {
try {
return htable().get(g);
} catch (IOException e) {
reset();
throw e;
}
}
public HTableDescriptor getHtableDescriptor() throws IOException {
try {
return htable().getTableDescriptor();
} catch (IOException e) {
reset();
throw e;
}
}
@Override
public void start() {
if (getCreate()) {
int attempt = 1;
while (true) {
try {
HBaseAdmin admin = hbase.admin();
if (!admin.tableExists(htableName)) {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(htableName));
logger.info("hbase table '" + htableName + "' does not exist, creating it... " + desc);
admin.createTable(desc);
}
break;
} catch (IOException e) {
logger.log(Level.WARNING, "(attempt " + attempt + ") problem creating hbase table " + htableName, e);
attempt++;
reset();
// back off up to 60 seconds between retries
try {
Thread.sleep(Math.min(attempt * 1000, 60000));
} catch (InterruptedException e1) {
}
}
}
}
super.start();
}
protected void reset() {
if (htable.get() != null) {
try {
htable.get().close();
} catch (IOException e) {
logger.log(Level.WARNING, "htablename='" + htableName + "' htable.close() threw " + e, e);
}
htable.remove();
}
if (hconn != null) {
try {
hconn.close();
} catch (IOException e) {
logger.log(Level.WARNING, "hconn.close() threw " + e, e);
}
// HConnectionManager.deleteStaleConnection(hconn);
hconn = null;
}
hbase.reset();
}
@Override
public synchronized void stop() {
super.stop();
reset();
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/HBaseTableBean.java
-78
View File
@@ -1,78 +0,0 @@
package org.archive.modules.recrawl.hbase;
import java.io.IOException;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.recrawl.PersistOnlineProcessor;
import org.springframework.context.Lifecycle;
/**
* base class for different types of HBaseTable Spring bean implementations.
* @author kenji
* @author nlevitt
*
*/
public abstract class HBaseTableBean implements Lifecycle {
protected String htableName = PersistOnlineProcessor.URI_HISTORY_DBNAME;
protected HBase hbase = new HBase();
protected transient boolean isRunning = false;
// <backward-compatibility>
public void setName(String name) {
this.htableName = name;
}
public String getName() {
return htableName;
}
// </backward-compatibility>
/**
* set name of single HTable this instance accesses.
* @param htableName
*/
public void setHtableName(String htableName) {
this.htableName = htableName;
}
public String getHtableName() {
return htableName;
}
public HBaseTableBean() {
super();
}
public void setHbase(HBase hbase) {
this.hbase = hbase;
}
public HBase getHbase() {
return hbase;
}
public abstract void put(Put p) throws IOException;
public abstract Result get(Get g) throws IOException;
public abstract HTableDescriptor getHtableDescriptor() throws IOException;
@Override
public boolean isRunning() {
return isRunning;
}
@Override
public void start() {
isRunning = true;
}
@Override
public synchronized void stop() {
isRunning = false;
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/MultiColumnRecrawlDataSchema.java
-140
View File
@@ -1,140 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
/**
* RecrawlDataSchema that stores each recrawl data properties in a separate column in single column
* family, whose name may be configured with {@link #setColumnFamily(String)} (default "f").
* <ul>
* <li>{@code s}: fetch status (as integer text)</li>
* <li>{@code d}: content digest (with {@code sha1:} prefix, Base32 text)</li>
* <li>{@code e}: ETag (enclosing quotes stripped)</li>
* <li>{@code m}: last-modified date-time (as integer timestamp, binary format)</li>
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
* </ul>
*
* @author kenji
*/
public class MultiColumnRecrawlDataSchema extends RecrawlDataSchemaBase implements RecrawlDataSchema, RecrawlAttributeConstants {
static final Logger logger = Logger.getLogger(MultiColumnRecrawlDataSchema.class.getName());
public static final byte[] COLUMN_STATUS = Bytes.toBytes("s");
public static final byte[] COLUMN_CONTENT_DIGEST = Bytes.toBytes("d");
public static final byte[] COLUMN_ETAG = Bytes.toBytes("e");
public static final byte[] COLUMN_LAST_MODIFIED = Bytes.toBytes("m");
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut()
*/
public Put createPut(CrawlURI uri) {
byte[] uriBytes = rowKeyForURI(uri);
byte[] key = uriBytes;
Put p = new Put(key);
String digest = uri.getContentDigestSchemeString();
if (digest != null) {
p.add(columnFamily, COLUMN_CONTENT_DIGEST, Bytes.toBytes(digest));
}
p.add(columnFamily, COLUMN_STATUS, Bytes.toBytes(Integer.toString(uri.getFetchStatus())));
if (uri.isHttpTransaction()) {
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
if (etag != null) {
// Etqg is usually quoted
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
etag = etag.substring(1, etag.length() - 1);
p.add(columnFamily, COLUMN_ETAG, Bytes.toBytes(etag));
}
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
if (lastmod != null) {
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
if (lastmod_sec == 0) {
try {
lastmod_sec = uri.getFetchCompletedTime();
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
if (lastmod_sec != 0)
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(lastmod_sec));
} else {
try {
long completed = uri.getFetchCompletedTime();
if (completed != 0)
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(completed));
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
}
return p;
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(java.util.Map, org.apache.hadoop.hbase.client.Result)
*/
public void load(Result result, CrawlURI curi) {
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
// URL.
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
if (nocrawl != null && nocrawl.length > 0) {
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
// is primarily intended for preventing crawler from stepping on traps.
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
curi.getAnnotations().add("nocrawl");
return;
}
// all column should have identical timestamp.
KeyValue rkv = result.getColumnLatest(columnFamily, COLUMN_STATUS);
long timestamp = rkv.getTimestamp();
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
// FetchHTTP ignores history with status <= 0
byte[] status = result.getValue(columnFamily, COLUMN_STATUS);
if (status != null) {
// Note that status is stored as integer text. It's typically three-chars
// that is less than 4-byte integer bits.
history.put(RecrawlAttributeConstants.A_STATUS, Integer.parseInt(Bytes.toString(status)));
byte[] etag = result.getValue(columnFamily, COLUMN_ETAG);
if (etag != null) {
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, Bytes.toString(etag));
}
byte[] lastmod = result.getValue(columnFamily, COLUMN_LAST_MODIFIED);
if (lastmod != null) {
long lastmod_sec = Bytes.toLong(lastmod);
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod_sec));
}
byte[] digest = result.getValue(columnFamily, COLUMN_CONTENT_DIGEST);
if (digest != null) {
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, Bytes.toString(digest));
}
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchema.java
-35
View File
@@ -1,35 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.archive.modules.CrawlURI;
/**
* @author kenji
*/
public interface RecrawlDataSchema {
public String getColumnFamily();
public Put createPut(CrawlURI uri);
public void load(Result result, CrawlURI curi);
// TODO: drop this method by revising createPut(CrawlURI) method.
public byte[] rowKeyForURI(CrawlURI curi);
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/RecrawlDataSchemaBase.java
-123
View File
@@ -1,123 +0,0 @@
package org.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Logger;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.canonicalize.CanonicalizationRule;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.FetchHistoryProcessor;
import org.archive.modules.recrawl.PersistProcessor;
/**
* implements common utility methods for implementing {@link RecrawlDataSchema}.
* <ul>
* <li>configuring single column family name</li>
* <li>formatting/parsing HTTP date text</li>
* <li>constructing row key</li>
* <li>preparing fetch-history array</li>
* </ul>
* @author kenji
*/
abstract public class RecrawlDataSchemaBase implements RecrawlDataSchema {
private static final Logger logger = Logger.getLogger(RecrawlDataSchemaBase.class.getName());
/**
* default value for {@link #columnFamily}.
*/
public static final byte[] DEFAULT_COLUMN_FAMILY = Bytes.toBytes("f");
protected byte[] columnFamily = DEFAULT_COLUMN_FAMILY;
public static final byte[] COLUMN_NOCRAWL = Bytes.toBytes("z");
/**
* default value for {@link #useCanonicalString}.
*/
public static boolean DEFAULT_USE_CANONICAL_STRING = true;
private boolean useCanonicalString = DEFAULT_USE_CANONICAL_STRING;
private CanonicalizationRule keyRule = null;
protected int historyLength = 2;
public RecrawlDataSchemaBase() {
super();
}
public void setColumnFamily(String colf) {
columnFamily = Bytes.toBytes(colf);
}
public boolean isUseCanonicalString() {
return useCanonicalString;
}
/**
* if set to true, canonicalized string will be used as row key, rather than URI
* @param useCanonicalString
*/
public void setUseCanonicalString(boolean useCanonicalString) {
this.useCanonicalString = useCanonicalString;
}
public String getColumnFamily() {
return Bytes.toString(columnFamily);
}
public CanonicalizationRule getKeyRule() {
return keyRule;
}
/**
* alternative canonicalization rule for generating row key from URI.
* TODO: currently unused.
* @param keyRule
*/
public void setKeyRule(CanonicalizationRule keyRule) {
this.keyRule = keyRule;
}
public int getHistoryLength() {
return historyLength;
}
/**
* maximum number of crawl history entries to retain in {@link CrawlURI}.
* when more than this number of crawl history entry is being added by
* {@link #getFetchHistory(CrawlURI, long)}, oldest entry will be discarded.
* {@code historyLength} should be the same number as
* {@link FetchHistoryProcessor#setHistoryLength(int)}, or FetchHistoryProcessor will
* reallocate the crawl history array.
* @param historyLength
* @see FetchHistoryProcessor#setHistoryLength(int)
*/
public void setHistoryLength(int historyLength) {
this.historyLength = historyLength;
}
/**
* calls {@link FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)} with {@link #historyLength}.
* @param uri CrawlURI from which fetch history is obtained.
* @return Map object for storing re-crawl data (never null).
* @see FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)
* @see FetchHistoryProcessor
*/
protected Map<String, Object> getFetchHistory(CrawlURI uri, long timestamp) {
return FetchHistoryHelper.getFetchHistory(uri, timestamp, historyLength);
}
/**
* return row key for {@code curi}.
* TODO: move this to HBasePersistProcessor by redesigning {@link RecrawlDataSchema}.
* @param curi {@link CrawlURI} for which a row is being fetched.
* @return row key
*/
public byte[] rowKeyForURI(CrawlURI curi) {
if (useCanonicalString) {
// TODO: use keyRule if specified.
return Bytes.toBytes(PersistProcessor.persistKeyFor(curi));
} else {
return Bytes.toBytes(curi.toString());
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleColumnJsonRecrawlDataSchema.java
-174
View File
@@ -1,174 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStatusCodes;
import org.archive.modules.recrawl.FetchHistoryHelper;
import org.archive.modules.recrawl.RecrawlAttributeConstants;
import org.json.JSONException;
import org.json.JSONObject;
/**
* {@linkplain SingleColumnJsonRecrawlDataSchema} stores all re-crawl data properties in a single column,
* in JSON format. As HBase stores each column paired with the row key, it takes a lot of space to store
* each re-crawl data property in its own column.
* <ul>
* <li>{@code r}: re-crawl data in JSON format</li>
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
* </ul>
* @author Kenji Nagahashi
*/
public class SingleColumnJsonRecrawlDataSchema extends RecrawlDataSchemaBase
implements RecrawlDataSchema {
static final Logger logger = Logger.getLogger(SingleColumnJsonRecrawlDataSchema.class.getName());
public static byte[] DEFAULT_COLUMN = Bytes.toBytes("r");
// JSON property names for re-crawl data properties
public static final String PROPERTY_STATUS = "s";
public static final String PROPERTY_CONTENT_DIGEST = "d";
public static final String PROPERTY_ETAG = "e";
public static final String PROPERTY_LAST_MODIFIED = "m";
// SHA1 scheme is assumed.
public static final String CONTENT_DIGEST_SCHEME = "sha1:";
// single column for storing JSON of re-crawl data
protected byte[] column = DEFAULT_COLUMN;
public void setColumn(String column) {
this.column = Bytes.toBytes(column);
}
public String getColumn() {
return Bytes.toString(column);
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut(org.archive.modules.CrawlURI)
*/
public Put createPut(CrawlURI uri) {
byte[] key = rowKeyForURI(uri);
Put p = new Put(key);
JSONObject jo = new JSONObject();
try {
// TODO should we post warning message when scheme != "sha1"?
String digest = uri.getContentDigestString();
if (digest != null) {
jo.put(PROPERTY_CONTENT_DIGEST, digest);
}
jo.put(PROPERTY_STATUS, uri.getFetchStatus());
if (uri.isHttpTransaction()) {
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
if (etag != null) {
// Etag is usually quoted
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
etag = etag.substring(1, etag.length() - 1);
jo.put(PROPERTY_ETAG, etag);
}
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
if (lastmod != null) {
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
if (lastmod_sec == 0) {
try {
lastmod_sec = uri.getFetchCompletedTime();
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
} else {
try {
long completed = uri.getFetchCompletedTime();
if (completed != 0)
jo.put(PROPERTY_LAST_MODIFIED, completed);
} catch (NullPointerException ex) {
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
}
}
}
} catch (JSONException ex) {
// should not happen - all values are either primitive or String.
logger.log(Level.SEVERE, "JSON translation failed", ex);
}
p.add(columnFamily, column, Bytes.toBytes(jo.toString()));
return p;
}
/* (non-Javadoc)
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(org.apache.hadoop.hbase.client.Result)
*/
public void load(Result result, CrawlURI curi) {
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
// URL.
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
if (nocrawl != null && nocrawl.length > 0) {
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
// is primarily intended for preventing crawler from stepping on traps.
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
curi.getAnnotations().add("nocrawl");
return;
}
KeyValue rkv = result.getColumnLatest(columnFamily, column);
long timestamp = rkv.getTimestamp();
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
if (history == null) {
// crawl history array is fully occupied by crawl history entries
// newer than timestamp.
return;
}
byte[] jsonBytes = rkv.getValue();
if (jsonBytes != null) {
JSONObject jo = null;
try {
jo = new JSONObject(Bytes.toString(jsonBytes));
} catch (JSONException ex) {
logger.warning(String.format("JSON parsing failed for key %1s: %2s",
result.getRow(), ex.getMessage()));
}
if (jo != null) {
int status = jo.optInt(PROPERTY_STATUS, -1);
if (status >= 0) {
history.put(RecrawlAttributeConstants.A_STATUS, status);
}
String digest = jo.optString(PROPERTY_CONTENT_DIGEST);
if (digest != null) {
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, CONTENT_DIGEST_SCHEME + digest);
}
String etag = jo.optString(PROPERTY_ETAG);
if (etag != null) {
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, etag);
}
long lastmod = jo.optLong(PROPERTY_LAST_MODIFIED);
if (lastmod > 0) {
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod));
}
}
}
}
}
contrib/src/main/java/org/archive/modules/recrawl/hbase/SingleHBaseTable.java
-380
View File
@@ -1,380 +0,0 @@
/*
* This file is part of the Heritrix web crawler (crawler.archive.org).
*
* Licensed to the Internet Archive (IA) by one or more individual
* contributors.
*
* The IA 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.archive.modules.recrawl.hbase;
import java.io.IOException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
/**
* simple HTable wrapper that shares single instance of HTable among threads.
* If you only perform get on HTable, this implementation
* should be good enough. If multiple threads performs Put, {@link HBaseTable}
* would be more efficient.
* <p>when HBase I/O fails due to issue with network/region server/zookeeper, this
* class waits for preset time (see {@link #setReconnectInterval(int)})
* before trying to reestablish HBase connection. During this hold-ff period, all
* {@link #get(Get)} and {@link #put(Put)} calls will fail.
*
* @author kenji
*/
public class SingleHBaseTable extends HBaseTableBean {
private static final Log LOG = LogFactory.getLog(SingleHBaseTable.class);
private HTableInterface table;
private volatile long tableError;
private ReentrantReadWriteLock tableUseLock = new ReentrantReadWriteLock();
boolean autoReconnect = true;
public boolean isAutoReconnect() {
return autoReconnect;
}
/**
* if set to {@code true}, HBaseClient tries to reconnect to the HBase master
* immediately when Put request failed due to connection loss (note {@link #put(Put)}
* still throws IOException even if autoReconnect is enabled.)
* @param autoReconnect true to enable auto-reconnect
*/
public void setAutoReconnect(boolean autoReconnect) {
this.autoReconnect = autoReconnect;
}
protected boolean autoFlush = true;
/**
* passed on to HTable's autoFlush property upon creation.
* @return true for enabling auto-flush.
*/
public boolean isAutoFlush() {
return autoFlush;
}
public void setAutoFlush(boolean autoFlush) {
this.autoFlush = autoFlush;
}
// default 3 minutes
private int reconnectInterval = 1000 * 3 * 60;
public int getReconnectInterval() {
return reconnectInterval;
}
/**
* set hold-off interval upon communication errors.
* @param reconnectInterval hold-off interval in milliseconds.
*/
public void setReconnectInterval(int reconnectInterval) {
this.reconnectInterval = reconnectInterval;
}
// counters
protected AtomicLong getCount = new AtomicLong();
// count of GET/PUT failures (i.e. not counting connection failures).
protected AtomicLong getErrorCount = new AtomicLong();
protected AtomicLong getSkipCount = new AtomicLong();
protected AtomicLong putCount = new AtomicLong();
protected AtomicLong putErrorCount = new AtomicLong();
protected AtomicLong putSkipCount = new AtomicLong();
protected AtomicLong connectCount = new AtomicLong();
public long getGetCount() { return getCount.get(); }
public long getGetErrorCount() { return getErrorCount.get(); }
public long getGetSkipCount() { return getSkipCount.get(); }
public long getPutCount() { return putCount.get(); }
public long getConnectCount() { return connectCount.get(); }
// for diagnosing deadlock situation
public Map<String, Object> getTableLockState() {
Map<String, Object> m = new LinkedHashMap<String, Object>();
m.put("readLockCount", tableUseLock.getReadLockCount());
m.put("queueLength", tableUseLock.getQueueLength());
m.put("writeLocked", tableUseLock.isWriteLocked());
return m;
}
public SingleHBaseTable() {
}
/**
* attempts to reconnect to HBase if table is null.
* must not be called with read-lock.
* @return existing or newly opened HTableInterface.
*/
protected HTableInterface getTable() {
if (table == null && autoReconnect)
openTable();
return table;
}
/**
* close HTable {@code table}, set current time to tableError if closing because
* of a communication error. should be called with write lock.
* @param htable HTable to close.
* @param byError true if closing because of an error.
*/
protected void closeTable(HTableInterface htable, boolean byError) {
if (htable == null) return;
if (table != htable) {
// other thread did closeTable on htable. don't close table.
return;
}
try {
table = null;
htable.close();
} catch (IOException ex) {
LOG.warn("error closing " + htable + " - some commits may have been lost");
}
if (byError) {
tableError = System.currentTimeMillis();
}
}
public void put(Put p) throws IOException {
putCount.incrementAndGet();
// trigger reconnection if necessary. as table can be modified before
// read lock is acquired, we don't read table variable here.
getTable();
boolean htableFailed = false;
HTableInterface htable = null;
Lock readLock = tableUseLock.readLock();
try {
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
putSkipCount.incrementAndGet();
throw new IOException("could not acquire read lock for HTable.");
}
} catch (InterruptedException ex) {
throw new IOException("interrupted while acquiring read lock", ex);
}
try {
htable = table;
if (htable == null) {
putSkipCount.incrementAndGet();
throw new IOException("HBase connection is unvailable.");
}
// HTable.put() buffers Puts and access to the buffer is not
// synchronized.
synchronized (htable) {
try {
htable.put(p);
} catch (NullPointerException ex) {
// HTable.put() throws NullPointerException when connection is lost.
// It is somewhat weird, so translate it to IOException.
putErrorCount.incrementAndGet();
htableFailed = true;
throw new IOException("hbase connection is lost", ex);
} catch (NotServingRegionException ex) {
putErrorCount.incrementAndGet();
// no need to close HTable.
throw ex;
} catch (IOException ex) {
putErrorCount.incrementAndGet();
htableFailed = true;
throw ex;
}
}
} finally {
readLock.unlock();
if (htableFailed) {
closeTable(htable, true);
}
}
}
public Result get(Get g) throws IOException {
getCount.incrementAndGet();
// trigger reconnection if necessary. as table can be modified before
// read lock is acquired, we don't read table variable here.
getTable();
boolean htableFailed = false;
HTableInterface htable = null;
Lock readLock = tableUseLock.readLock();
try {
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
getSkipCount.incrementAndGet();
throw new IOException("could not acquire read lock for HTable.");
}
} catch (InterruptedException ex) {
throw new IOException("interrupted while acquiring read lock", ex);
}
try {
htable = table;
if (htable == null) {
getSkipCount.incrementAndGet();
throw new IOException("HBase connection is unvailable.");
}
try {
return htable.get(g);
} catch (NotServingRegionException ex) {
// caused by disruption to HBase cluster. no need to
// refresh HBase connection, since connection itself
// is working okay.
// TODO: should we need to back-off for a while? other
// regions may still be accessible.
getErrorCount.incrementAndGet();
throw ex;
} catch (IOException ex) {
getErrorCount.incrementAndGet();
htableFailed = true;
throw ex;
}
} finally {
readLock.unlock();
if (htableFailed) {
closeTable(htable, true);
}
}
}
@Override
public HTableDescriptor getHtableDescriptor() throws IOException {
HTableInterface table = getTable();
if (table == null) {
throw new IOException("HBase connection is unavailable.");
}
return table.getTableDescriptor();
}
public boolean inBackoffPeriod() {
return (tableError > 0 &&
(System.currentTimeMillis() - tableError) < reconnectInterval);
}
/**
* timestamp of the last Put/Get error.
* @return timestamp in ms.
*/
public long getTableErrorTime() {
return tableError;
}
/**
* connect to HBase.
* it does nothing if table is non-null, or it is in the back-off period since
* the last error.
* should be called with write lock.
*/
protected boolean openTable() {
if (table != null) return true;
// fail immediately if we're in back-off period.
if (inBackoffPeriod()) return false;
try {
HTable t = new HTable(hbase.configuration(), Bytes.toBytes(htableName));
connectCount.incrementAndGet();
t.setAutoFlush(autoFlush);
table = t;
tableError = 0;
return true;
} catch (TableNotFoundException ex) {
// ex.getMessage() only has table name. be a little bit more friendly.
LOG.warn("failed to connect to HTable \"" + htableName + "\": Table Not Found");
tableError = System.currentTimeMillis();
return false;
} catch (IOException ex) {
LOG.warn("failed to connect to HTable \"" + htableName + "\" (" + ex.getMessage() + ")");
tableError = System.currentTimeMillis();
return false;
}
}
/**
* number of seconds to wait for acquiring read lock.
* if read lock is not acquired within this many seconds (probably
* due to deadlock situation on write-lock side), {@link #get(Get)} will
* <i>silently</i> fail.
*/
public final static long TRY_READ_LOCK_TIMEOUT = 5;
/**
* number of seconds to wait for acquiring write lock.
*/
public final static long TRY_WRITE_LOCK_TIMEOUT = 10;
/**
* close current connection and establish new connection.
* fails silently if back-off period is in effect.
*/
protected void reconnect(boolean onerror) throws IOException, InterruptedException {
// avoid deadlock situation caused by attempting
// to acquire write lock while holding read lock.
// there'd be no real dead-lock now that timeout on write lock is implemented,
// but it's nice to know there's a bug in locking.
if (tableUseLock.getReadHoldCount() > 0) {
LOG.warn("avoiding deadlock: reconnect() called by thread with read lock.");
return;
}
Lock writeLock = tableUseLock.writeLock();
if (!writeLock.tryLock(TRY_WRITE_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
LOG.warn("reconnect() could not acquire write lock on tableUseLock for " +
TRY_WRITE_LOCK_TIMEOUT + "s, giving up.");
return;
}
try {
closeTable(table, onerror);
openTable();
} finally {
writeLock.unlock();
}
}
/**
* close current connection and establish new connection.
* for refreshing stale connection through scripting.
* resets tableErrorTime to zero (it will be set to non-zero if
* reconnection attempt fails).
* @throws IOException
* @throws InterruptedException
*/
public void reconnect() throws IOException, InterruptedException {
tableError = 0;
reconnect(false);
}
// public boolean isRunning() {
// return table != null;
// }
public void start() {
super.start();
openTable();
}
public void stop() {
if (table != null) {
try {
table.close();
} catch (IOException ex) {
LOG.warn("table.close() failed", ex);
}
}
table = null;
super.stop();
}
}
engine/src/main/java/org/archive/crawler/restlet/JobResource.java
+6 -1
View File
@@ -224,7 +224,12 @@ public class JobResource extends BaseResource {
} else if ("pause".equals(action)) {
cj.getCrawlController().requestCrawlPause();
} else if ("unpause".equals(action)) {
cj.getCrawlController().requestCrawlResume();
try {
cj.getCrawlController().requestCrawlResume();
} catch (Exception e){
System.err.println(getName() + ": exception " + e + " during unpause.");
e.printStackTrace();
}
} else if ("checkpoint".equals(action)) {
String cp = cj.getCheckpointService().requestCrawlCheckpoint();
if (StringUtils.isNotEmpty(cp)) {
engine/src/main/java/org/archive/crawler/restlet/PagedRepresentation.java
-1
View File
@@ -144,7 +144,6 @@ public class PagedRepresentation extends CharacterRepresentation {
pw.println("</pre>");
emitControls(pw);
pw.close();
}
/**
engine/src/main/java/org/archive/crawler/restlet/RateLimitGuard.java
+1
View File
@@ -53,6 +53,7 @@ public class RateLimitGuard extends DigestAuthenticator {
long now = System.currentTimeMillis();
long sleepMs = (lastFailureTime+MIN_MS_BETWEEN_ATTEMPTS)-now;
if(sleepMs>0) {
System.out.println(new java.util.Date() + " " + getName() + ": trying to sleep");
try {
Thread.sleep(sleepMs);
} catch (InterruptedException e) {
engine/src/test/java/org/archive/crawler/util/BloomUriUniqFilterTest.java
+2 -3
View File
@@ -67,7 +67,7 @@ implements UriUniqFilter.CrawlUriReceiver {
this.filter.addForce(this.getUri(),
new CrawlURI(UURIFactory.getInstance(this.getUri())));
// Should only have add 'this' once.
assertTrue("Count is off", this.filter.count() == 1);
assertEquals("Count is off", 1, this.filter.count());
}
/**
@@ -104,8 +104,7 @@ implements UriUniqFilter.CrawlUriReceiver {
logger.fine("Readded subset " + list.size() + " in " +
(System.currentTimeMillis() - start));
assertTrue("Count is off: " + filter.count(),
filter.count() == MAX_COUNT);
assertEquals("Count is off", MAX_COUNT, filter.count());
}
public void testNote() {
modules/pom.xml
+2 -1
View File
@@ -1,5 +1,6 @@
<?xml version="1.0"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.archive</groupId>
<artifactId>heritrix</artifactId>
modules/src/main/java/org/archive/modules/CrawlURI.java
+1
View File
@@ -859,6 +859,7 @@ implements Reporter, Serializable, OverlayContext, Comparable<CrawlURI> {
this.httpRecorder = null;
this.fetchStatus = S_UNATTEMPTED;
this.setPrerequisite(false);
this.clearPrerequisiteUri();
this.contentSize = UNCALCULATED;
this.contentLength = UNCALCULATED;
// Clear 'links extracted' flag.
modules/src/main/java/org/archive/modules/extractor/ExtractorHTML.java
+2 -1
View File
@@ -981,7 +981,8 @@ public class ExtractorHTML extends ContentExtractor implements InitializingBean
} else if ("refresh".equalsIgnoreCase(httpEquiv) && content != null) {
int urlIndex = content.indexOf("=") + 1;
if(urlIndex>0) {
String refreshUri = content.substring(urlIndex);
// strip any quotes ("') characters from the URL value.
String refreshUri = TextUtils.replaceAll("[\"']", content.substring(urlIndex), "");
try {
int max = getExtractorParameters().getMaxOutlinks();
addRelativeToBase(curi, max, refreshUri,
modules/src/main/java/org/archive/modules/extractor/ExtractorMultipleRegex.java
+5
View File
@@ -192,6 +192,7 @@ public class ExtractorMultipleRegex extends Extractor {
while (matcher.find()) {
add(new GroupList(matcher));
}
TextUtils.recycleMatcher(matcher);
}
public MatchList(GroupList... groupList) {
for (GroupList x: groupList) {
@@ -219,6 +220,7 @@ public class ExtractorMultipleRegex extends Extractor {
matchLists = new LinkedHashMap<String,MatchList>();
matchLists.put("uriRegex", new MatchList(new GroupList(matcher)));
} else {
TextUtils.recycleMatcher(matcher);
return; // if uri regex doesn't match, we're done
}
@@ -229,6 +231,7 @@ public class ExtractorMultipleRegex extends Extractor {
curi.getNonFatalFailures().add(e);
LOGGER.log(Level.WARNING, "Failed get of replay char sequence in "
+ Thread.currentThread().getName(), e);
TextUtils.recycleMatcher(matcher);
return;
}
@@ -237,6 +240,7 @@ public class ExtractorMultipleRegex extends Extractor {
String regex = getContentRegexes().get(regexName);
MatchList matchList = new MatchList(regex, cs);
if (matchList.isEmpty()) {
TextUtils.recycleMatcher(matcher);
return; // no match found for regex, so we can stop now
}
matchLists.put(regexName, matchList);
@@ -257,6 +261,7 @@ public class ExtractorMultipleRegex extends Extractor {
Map<String, Object> bindings = makeBindings(matchLists, regexNames, i);
buildAndAddOutlink(curi, bindings);
}
TextUtils.recycleMatcher(matcher);
}
// bindings are the variables available to populate the template
modules/src/main/java/org/archive/modules/writer/ARCWriterProcessor.java
+2 -4
View File
@@ -129,8 +129,7 @@ public class ARCWriterProcessor extends WriterPoolProcessor {
// We just closed the file because it was larger than maxBytes.
// Add to the totalBytesWritten the size of the first record
// in the file, if any.
setTotalBytesWritten(getTotalBytesWritten() +
(writer.getPosition() - position));
addTotalBytesWritten(writer.getPosition() - position);
position = writer.getPosition();
}
@@ -155,8 +154,7 @@ public class ARCWriterProcessor extends WriterPoolProcessor {
throw e;
} finally {
if (writer != null) {
setTotalBytesWritten(getTotalBytesWritten() +
(writer.getPosition() - position));
addTotalBytesWritten(writer.getPosition() - position);
getPool().returnFile(writer);
String filename = writer.getFile().getName();
modules/src/main/java/org/archive/modules/writer/BaseWARCWriterProcessor.java
+1 -1
View File
@@ -211,7 +211,7 @@ abstract public class BaseWARCWriterProcessor extends WriterPoolProcessor
+ WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK)
+ " bytes to " + writer.getFile().getName() + " for " + curi);
}
setTotalBytesWritten(getTotalBytesWritten() + (writer.getPosition() - startPosition));
addTotalBytesWritten(writer.getPosition() - startPosition);
curi.addExtraInfo("warcFilename", writer.getFilenameWithoutOccupiedSuffix());
curi.addExtraInfo("warcFileOffset", startPosition);
modules/src/main/java/org/archive/modules/writer/WARCWriterChainProcessor.java
+56 -2
View File
@@ -1,12 +1,16 @@
package org.archive.modules.writer;
import java.io.InputStream;
import java.io.IOException;
import java.net.URI;
import java.util.Arrays;
import java.util.List;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.io.IOUtils;
import org.archive.io.warc.WARCRecordInfo;
import org.archive.io.warc.WARCWriter;
import org.archive.modules.CrawlURI;
@@ -22,6 +26,8 @@ import org.archive.modules.warc.RevisitRecordBuilder;
import org.archive.modules.warc.WARCRecordBuilder;
import org.archive.modules.warc.WhoisResponseRecordBuilder;
import org.archive.spring.HasKeyedProperties;
import org.json.JSONException;
import org.json.JSONObject;
/**
* WARC writer processor. The types of records that to be written can be
@@ -123,8 +129,7 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements
// We rolled over to a new warc and wrote a warcinfo record.
// Tally stats and reset temp stats, to avoid including warcinfo
// record in stats for current url.
setTotalBytesWritten(getTotalBytesWritten() +
(writer.getPosition() - position));
addTotalBytesWritten(writer.getPosition() - position);
addStats(writer.getTmpStats());
writer.resetTmpStats();
writer.resetTmpRecordLog();
@@ -159,6 +164,17 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements
WARCRecordInfo record = recordBuilder.buildRecord(curi, concurrentTo);
if (record != null) {
writer.writeRecord(record);
InputStream is = null;
try {
is = record.getContentStream();
is.close();
}
catch (Exception e){
logger.log(Level.WARNING, "problem closing Warc Record Content Stream " + e);
}
finally {
IOUtils.closeQuietly(record.getContentStream()); //Closing one way or the other seems to leave some file handles open. Calling close() and using closeQuietly() handles both FileStreams and FileChannels
}
if (concurrentTo == null) {
concurrentTo = record.getRecordId();
}
@@ -166,4 +182,42 @@ public class WARCWriterChainProcessor extends BaseWARCWriterProcessor implements
}
}
}
@Override
protected JSONObject toCheckpointJson() throws JSONException {
JSONObject json = super.toCheckpointJson();
json.put("urlsWritten", urlsWritten);
json.put("stats", stats);
return json;
}
@Override
protected void fromCheckpointJson(JSONObject json) throws JSONException {
super.fromCheckpointJson(json);
// conditionals below are for backward compatibility with old checkpoints
if (json.has("urlsWritten")) {
urlsWritten.set(json.getLong("urlsWritten"));
}
if (json.has("stats")) {
HashMap<String, Map<String, Long>> cpStats = new HashMap<String, Map<String, Long>>();
JSONObject jsonStats = json.getJSONObject("stats");
if (JSONObject.getNames(jsonStats) != null) {
for (String key1: JSONObject.getNames(jsonStats)) {
JSONObject jsonSubstats = jsonStats.getJSONObject(key1);
if (!cpStats.containsKey(key1)) {
cpStats.put(key1, new HashMap<String, Long>());
}
Map<String, Long> substats = cpStats.get(key1);
for (String key2: JSONObject.getNames(jsonSubstats)) {
long value = jsonSubstats.getLong(key2);
substats.put(key2, value);
}
}
addStats(cpStats);
}
}
}
}
modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java
+5 -2
View File
@@ -166,8 +166,7 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC
// We rolled over to a new warc and wrote a warcinfo record.
// Tally stats and reset temp stats, to avoid including warcinfo
// record in stats for current url.
setTotalBytesWritten(getTotalBytesWritten() +
(writer.getPosition() - position));
addTotalBytesWritten(writer.getPosition() - position);
addStats(writer.getTmpStats());
writer.resetTmpStats();
writer.resetTmpRecordLog();
@@ -647,6 +646,7 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC
protected JSONObject toCheckpointJson() throws JSONException {
JSONObject json = super.toCheckpointJson();
json.put("urlsWritten", urlsWritten);
json.put("totalBytesWritten", getTotalBytesWritten());
json.put("stats", stats);
return json;
}
@@ -660,6 +660,9 @@ public class WARCWriterProcessor extends BaseWARCWriterProcessor implements WARC
if (json.has("urlsWritten")) {
urlsWritten.set(json.getLong("urlsWritten"));
}
if (json.has("totalBytesWritten")) {
setTotalBytesWritten(json.getLong("totalBytesWritten"));
}
if (json.has("stats")) {
HashMap<String, Map<String, Long>> cpStats = new HashMap<String, Map<String, Long>>();
modules/src/main/java/org/archive/modules/writer/WriterPoolProcessor.java
+15 -4
View File
@@ -31,6 +31,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;
import org.archive.checkpointing.Checkpoint;
@@ -42,6 +43,7 @@ import org.archive.modules.CrawlMetadata;
import org.archive.modules.CrawlURI;
import org.archive.modules.ProcessResult;
import org.archive.modules.Processor;
import org.archive.modules.deciderules.DecideResult;
import org.archive.modules.deciderules.recrawl.IdenticalDigestDecideRule;
import org.archive.modules.net.CrawlHost;
import org.archive.modules.net.ServerCache;
@@ -273,7 +275,7 @@ implements Lifecycle, Checkpointable, WriterPoolSettings {
/**
* Total number of bytes written to disc.
*/
private long totalBytesWritten = 0;
private AtomicLong totalBytesWritten = new AtomicLong();
private AtomicInteger serial = new AtomicInteger();
@@ -315,7 +317,7 @@ implements Lifecycle, Checkpointable, WriterPoolSettings {
if (max <= 0) {
return ProcessResult.PROCEED;
}
if (max <= this.totalBytesWritten) {
if (max <= getTotalBytesWritten()) {
return ProcessResult.FINISH; // FIXME: Specify reason
// controller.requestCrawlStop(CrawlStatus.FINISHED_WRITE_LIMIT);
}
@@ -362,6 +364,12 @@ implements Lifecycle, Checkpointable, WriterPoolSettings {
return false;
}
if (getShouldProcessRule().decisionFor(curi) == DecideResult.REJECT) {
curi.getAnnotations().add(ANNOTATION_UNWRITTEN + ":rejected("
+ getShouldProcessRule().getClass() + ")");
return false;
}
return true;
}
@@ -435,11 +443,14 @@ implements Lifecycle, Checkpointable, WriterPoolSettings {
}
protected long getTotalBytesWritten() {
return totalBytesWritten;
return totalBytesWritten.get();
}
protected void setTotalBytesWritten(long totalBytesWritten) {
this.totalBytesWritten = totalBytesWritten;
this.totalBytesWritten.set(totalBytesWritten);
}
protected void addTotalBytesWritten(long bytesWritten) {
this.totalBytesWritten.addAndGet(bytesWritten);
}
public abstract List<String> getMetadata();