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

[HER-1789] stats from WARCWriterProcessor

Browse Source 
* WriterPoolMember.java
    write(*), copyFrom() - return number of (uncompressed) bytes written, for
    convenience
* WARCWriter.java
    tally by record type: number of records written, content bytes, total
    bytes, and (possibly compressed) size on disk
* WARCWriterProcessor.java
    keep totals and write report to processors-report.txt
This commit is contained in:
nlevitt
2011-01-12 00:54:01 +00:00
parent 5fe28f2ecc
commit ef9d2e8c0e
3 changed files with 206 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/src/main/java/org/archive/io/WriterPoolMember.java
+28 -13
View File
@@ -406,21 +406,33 @@ public abstract class WriterPoolMember implements ArchiveFileConstants {
return compressed;
}
protected void write(final byte [] b) throws IOException {
this.out.write(b);
/**
* @return number of bytes written, which is always {@code b.length}
*/
protected int write(final byte[] b) throws IOException {
this.out.write(b);
return b.length;
}
protected void flush() throws IOException {
this.out.flush();
}
protected void write(byte[] b, int off, int len) throws IOException {
this.out.write(b, off, len);
}
protected void flush() throws IOException {
this.out.flush();
}
protected void write(int b) throws IOException {
this.out.write(b);
}
/**
* @return number of bytes written, which is always {@code len}
*/
protected int write(byte[] b, int off, int len) throws IOException {
this.out.write(b, off, len);
return len;
}
/**
* @return number of bytes written, which is always {@code 1}
*/
protected int write(int b) throws IOException {
this.out.write(b);
return 1;
}
/**
* Copy bytes from the provided InputStream to the target file/stream being
@@ -433,9 +445,10 @@ public abstract class WriterPoolMember implements ArchiveFileConstants {
* @param enforceLength
* whether to throw an exception if too many/too few bytes are
* available from stream
* @return number of bytes written (normally equal to {@code enforceLength})
* @throws IOException
*/
protected void copyFrom(final InputStream is, final long recordLength,
protected long copyFrom(final InputStream is, final long recordLength,
boolean enforceLength) throws IOException {
int read = scratchbuffer.length;
long tot = 0;
@@ -452,6 +465,8 @@ public abstract class WriterPoolMember implements ArchiveFileConstants {
throw new IOException("Read " + tot + " but expected "
+ recordLength);
}
return tot;
}
public void close() throws IOException {
commons/src/main/java/org/archive/io/warc/WARCWriter.java
+121 -28
View File
@@ -27,6 +27,7 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -56,6 +57,12 @@ import org.archive.util.anvl.Element;
*/
public class WARCWriter extends WriterPoolMember
implements WARCConstants {
public static final String TOTALS = "totals";
public static final String SIZE_ON_DISK = "sizeOnDisk";
public static final String TOTAL_BYTES = "totalBytes";
public static final String CONTENT_BYTES = "contentBytes";
public static final String NUM_RECORDS = "numRecords";
private static final Logger logger =
Logger.getLogger(WARCWriter.class.getName());
@@ -77,6 +84,7 @@ implements WARCConstants {
*/
private final List<String> fileMetadata;
private Map<String,Map<String,Long>> stats;
/**
* Shutdown Constructor
@@ -226,51 +234,114 @@ implements WARCConstants {
}
protected void writeRecord(final String type, final String url,
final String create14DigitDate, final String mimetype,
final URI recordId, ANVLRecord xtraHeaders,
final String create14DigitDate, final String mimetype,
final URI recordId, ANVLRecord xtraHeaders,
final InputStream contentStream, final long contentLength,
boolean enforceLength)
throws IOException {
if (!TYPES_LIST.contains(type)) {
throw new IllegalArgumentException("Unknown record type: " + type);
}
if (contentLength == 0 &&
if (!TYPES_LIST.contains(type)) {
throw new IllegalArgumentException("Unknown record type: " + type);
}
if (contentLength == 0 &&
(xtraHeaders == null || xtraHeaders.size() <= 0)) {
throw new IllegalArgumentException("Cannot write record " +
"of content-length zero and base headers only.");
}
throw new IllegalArgumentException("Cannot write record " +
"of content-length zero and base headers only.");
}
String header;
try {
header = createRecordHeader(type, url,
create14DigitDate, mimetype, recordId, xtraHeaders,
contentLength);
String header;
try {
header = createRecordHeader(type, url,
create14DigitDate, mimetype, recordId, xtraHeaders,
contentLength);
} catch (IllegalArgumentException e) {
logger.log(Level.SEVERE,"could not write record type: " + type
+ "for URL: " + url, e);
return;
}
try {
} catch (IllegalArgumentException e) {
logger.log(Level.SEVERE,"could not write record type: " + type
+ "for URL: " + url, e);
return;
}
long contentBytes = 0;
long totalBytes = 0;
long startPosition;
try {
checkSize(); // may start a new output file
startPosition = getPosition();
preWriteRecordTasks();
// TODO: Revisit endcoding of header.
write(header.getBytes(WARC_HEADER_ENCODING));
totalBytes += write(header.getBytes(WARC_HEADER_ENCODING));
if (contentStream != null && contentLength > 0) {
// Write out the header/body separator.
write(CRLF_BYTES); // TODO: should this be written even for zero-length?
copyFrom(contentStream, contentLength, enforceLength);
totalBytes += write(CRLF_BYTES); // TODO: should this be written even for zero-length?
contentBytes += copyFrom(contentStream, contentLength, enforceLength);
totalBytes += contentBytes;
}
// Write out the two blank lines at end of all records.
write(CRLF_BYTES);
write(CRLF_BYTES);
totalBytes += write(CRLF_BYTES);
totalBytes += write(CRLF_BYTES);
} finally {
postWriteRecordTasks();
}
// TODO: should this be in the finally block?
tally(type, contentBytes, totalBytes, getPosition() - startPosition);
}
// if compression is enabled, sizeOnDisk means compressed bytes; if not, it
// should be the same as totalBytes (right?)
protected void tally(String recordType, long contentBytes, long totalBytes, long sizeOnDisk) {
if (stats == null) {
stats = new HashMap<String,Map<String,Long>>();
}
// add to stats for this record type
Map<String,Long> substats = stats.get(recordType);
if (substats == null) {
substats = new HashMap<String,Long>();
stats.put(recordType, substats);
}
subtally(substats, contentBytes, totalBytes, sizeOnDisk);
// add to totals
substats = stats.get(TOTALS);
if (substats == null) {
substats = new HashMap<String,Long>();
stats.put(TOTALS, substats);
}
subtally(substats, contentBytes, totalBytes, sizeOnDisk);
}
protected void subtally(Map<String,Long> substats, long contentBytes,
long totalBytes, long sizeOnDisk) {
if (substats.get(NUM_RECORDS) == null) {
substats.put(NUM_RECORDS, 1l);
} else {
substats.put(NUM_RECORDS, substats.get(CONTENT_BYTES) + 1l);
}
if (substats.get(CONTENT_BYTES) == null) {
substats.put(CONTENT_BYTES, contentBytes);
} else {
substats.put(CONTENT_BYTES, substats.get(CONTENT_BYTES) + contentBytes);
}
if (substats.get(TOTAL_BYTES) == null) {
substats.put(TOTAL_BYTES, totalBytes);
} else {
substats.put(TOTAL_BYTES, substats.get(TOTAL_BYTES) + totalBytes);
}
if (substats.get(SIZE_ON_DISK) == null) {
substats.put(SIZE_ON_DISK, sizeOnDisk);
} else {
substats.put(SIZE_ON_DISK, substats.get(SIZE_ON_DISK) + sizeOnDisk);
}
}
protected URI generateRecordId(final Map<String, String> qualifiers)
throws IOException {
URI rid = null;
@@ -453,4 +524,26 @@ implements WARCConstants {
}
return result;
}
public void resetStats() {
if (stats != null) {
for (Map<String,Long> substats : stats.values()) {
for (Map.Entry<String,Long> entry : substats.entrySet()) {
entry.setValue(0l);
}
}
}
}
public Map<String,Map<String,Long>> getStats() {
return stats;
}
public static long getStat(Map<String,Map<String,Long>> statz, String key, String subkey) {
if (statz != null && statz.get(key) != null && statz.get(key).get(subkey) != null) {
return statz.get(key).get(subkey);
} else {
return 0;
}
}
}
modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java
+57 -16
View File
@@ -94,6 +94,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor {
private static final long serialVersionUID = 6182850087635847443L;
private static final Logger logger =
Logger.getLogger(WARCWriterProcessor.class.getName());
private HashMap<String,Map<String,Long>> stats;
private int urlsWritten;
public long getDefaultMaxFileSize() {
return 1000000000L; // 1 SI giga-byte (10^9 bytes), per WARC appendix A
@@ -200,21 +203,10 @@ public class WARCWriterProcessor extends WriterPoolProcessor {
final CrawlURI curi)
throws IOException {
WriterPoolMember writer = getPool().borrowFile();
long position = writer.getPosition();
// See if we need to open a new file because we've exceeed maxBytes.
// Call to checkFileSize will open new file if we're at maximum for
// current file.
writer.checkSize();
if (writer.getPosition() != position) {
// 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));
position = writer.getPosition();
}
WARCWriter w = (WARCWriter)writer;
w.resetStats();
try {
// Write a request, response, and metadata all in the one
// 'transaction'.
@@ -240,14 +232,37 @@ public class WARCWriterProcessor extends WriterPoolProcessor {
throw e;
} finally {
if (writer != null) {
setTotalBytesWritten(getTotalBytesWritten() +
(writer.getPosition() - position));
if (WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0) {
addStats(w.getStats());
urlsWritten++;
}
logger.fine("wrote " + WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK) + " bytes to " + w.getFile().getName() + " for " + curi);
setTotalBytesWritten(getTotalBytesWritten() + WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK));
getPool().returnFile(writer);
}
}
return checkBytesWritten();
}
protected void addStats(Map<String,Map<String,Long>> statz) {
if (stats == null) {
stats = new HashMap<String,Map<String,Long>>();
}
for (String key: statz.keySet()) {
if (stats.get(key) == null) {
stats.put(key, new HashMap<String,Long>());
}
for (String subkey: statz.get(key).keySet()) {
if (stats.get(key).get(subkey) == null) {
stats.get(key).put(subkey, statz.get(key).get(subkey));
} else {
stats.get(key).put(subkey, stats.get(key).get(subkey) + statz.get(key).get(subkey));
}
}
}
}
private void writeDnsRecords(final CrawlURI curi, WARCWriter w,
final URI baseid, final String timestamp) throws IOException {
ANVLRecord headers = null;
@@ -648,4 +663,30 @@ public class WARCWriterProcessor extends WriterPoolProcessor {
record.addLabelValue(label, value);
}
}
@Override
public String report() {
logger.info("final stats: " + stats);
StringBuilder buf = new StringBuilder();
buf.append("Processor: " + getClass().getName() + "\n");
buf.append(" Function: Writes WARCs\n");
buf.append(" Total CrawlURIs: " + urlsWritten + "\n");
buf.append(" Revisit records: " + WARCWriter.getStat(stats, WARCWriter.REVISIT, WARCWriter.NUM_RECORDS) + "\n");
long bytes = WARCWriter.getStat(stats, WARCWriter.RESPONSE, WARCWriter.CONTENT_BYTES)
+ WARCWriter.getStat(stats, WARCWriter.RESOURCE, WARCWriter.CONTENT_BYTES);
buf.append(" Crawled content bytes (including http headers): "
+ bytes + " (" + ArchiveUtils.formatBytesForDisplay(bytes) + ")\n");
bytes = WARCWriter.getStat(stats, WARCWriter.TOTALS, WARCWriter.TOTAL_BYTES);
buf.append(" Total uncompressed bytes (including all warc records): "
+ bytes + " (" + ArchiveUtils.formatBytesForDisplay(bytes) + ")\n");
buf.append(" Total size on disk ("+ (getCompress() ? "compressed" : "uncompressed") + "): "
+ getTotalBytesWritten() + " (" + ArchiveUtils.formatBytesForDisplay(getTotalBytesWritten()) + ")\n");
return buf.toString();
}
}