Revision: 16612
http://sourceforge.net/p/gate/code/16612
Author: valyt
Date: 2013-03-22 10:27:20 +0000 (Fri, 22 Mar 2013)
Log Message:
-----------
- more work on the ARC enumerator / naming strategy / input handler triad to
make it work with the new style document IDs.
- (untested) implementation for reading ARC records from a remote URL, using
byte ranges.
- removed some obsolete code.
Modified Paths:
--------------
gcp/trunk/src/gate/cloud/io/arc/ARCDocumentEnumerator.java
gcp/trunk/src/gate/cloud/io/arc/ARCDocumentNamingStrategy.java
gcp/trunk/src/gate/cloud/io/arc/ARCInputHandler.java
gcp/trunk/src/gate/cloud/io/file/SimpleNamingStrategy.java
Modified: gcp/trunk/src/gate/cloud/io/arc/ARCDocumentEnumerator.java
===================================================================
--- gcp/trunk/src/gate/cloud/io/arc/ARCDocumentEnumerator.java 2013-03-22
02:43:36 UTC (rev 16611)
+++ gcp/trunk/src/gate/cloud/io/arc/ARCDocumentEnumerator.java 2013-03-22
10:27:20 UTC (rev 16612)
@@ -218,12 +218,14 @@
if(mimeTypes == null || mimeTypes.contains(header.getMimetype())) {
// found a good document
Object2ObjectArrayMap<String, String> attrs =
- new Object2ObjectArrayMap<String, String>(new String[2],
- new String[2]);
+ new Object2ObjectArrayMap<String, String>(new String[3],
+ new String[3]);
attrs.put(ARCInputHandler.RECORD_OFFSET_ATTR,
Long.toString(header.getOffset()));
attrs.put(ARCInputHandler.RECORD_LENGTH_ATTR,
Long.toString(recordLength));
+ attrs.put(ARCInputHandler.RECORD_POSITION_ATTR,
+ Long.toString(inputSequence));
next = new DocumentID(header.getUrl(), attrs);
return;
}
Modified: gcp/trunk/src/gate/cloud/io/arc/ARCDocumentNamingStrategy.java
===================================================================
--- gcp/trunk/src/gate/cloud/io/arc/ARCDocumentNamingStrategy.java
2013-03-22 02:43:36 UTC (rev 16611)
+++ gcp/trunk/src/gate/cloud/io/arc/ARCDocumentNamingStrategy.java
2013-03-22 10:27:20 UTC (rev 16612)
@@ -18,6 +18,9 @@
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.log4j.Logger;
+
+import gate.cloud.batch.DocumentID;
import gate.cloud.io.file.SimpleNamingStrategy;
import gate.util.GateException;
@@ -25,13 +28,16 @@
* A naming strategy to convert document IDs suitable for use with
* an {@link ARCInputHandler} to file paths suitable for saving the
* results of their processing. It assumes that the document IDs
- * start with one or more digits, and arranges the output files under
- * a single document root in numbered directories constructed by padding
- * the document number to the left with zeros and creating intermediate
- * directories according to a configurable pattern. The default
- * pattern is '3/3', which pads the numbers to a minimum of 6 digits
- * and then splits them up into groups of three. The remainder of the
- * ID after the number is cleaned up to remove any URL protocol like
+ * use the record URL as the id text (see {@link DocumentID#getIdText()}), and
+ * that the document ID attributes (see {@link DocumentID#getAttributes()})
+ * include a numeric value for {@link ARCInputHandler#RECORD_POSITION_ATTR}
that
+ * is used as a sequence number.
+ * It arranges the output files under a single document root in numbered
+ * directories constructed by padding the document sequence number to the left
+ * with zeros and creating intermediate directories according to a configurable
+ * pattern. The default pattern is '3/3', which pads the numbers to a minimum
+ * of 6 digits and then splits them up into groups of three. The remainder of
+ * the ID after the number is cleaned up to remove any URL protocol like
* http:// and any query string or fragment. Any sequences of non-ASCII
* characters are removed and any remaining slashes or colons are replaced
* with underscores. For example with the default pattern, the document
@@ -44,20 +50,14 @@
*
*/
public class ARCDocumentNamingStrategy extends SimpleNamingStrategy {
+
+ private static final Logger logger =
Logger.getLogger(ARCDocumentEnumerator.class);
+
protected String pattern;
protected long[] divisors;
/**
- * Regular expression used to parse the document IDs. It extracts
- * digits at the beginning, then ignores anything up to the first
- * ://, then extracts everything between that and the first question
- * mark or hash, if any, or the end of the string otherwise.
- */
- protected static final Pattern ID_PATTERN = Pattern.compile(
- "^(\\d+)_*(?:.*?://)?(.*?)(?:\\?.*)?(?:#.*)?$");
-
- /**
* A pattern matching any sequence of non-ascii characters.
*/
protected static final Pattern NON_ASCII_PATTERN = Pattern.compile(
@@ -112,14 +112,25 @@
pattern = patternBuilder.toString();
}
- protected String relativePathFor(String id) {
+ @Override
+ protected String relativePathFor(DocumentID id) {
StringBuffer pathBuilder = new StringBuffer();
- Matcher idMatcher = ID_PATTERN.matcher(id);
- if(!idMatcher.find()) {
- throw new IllegalArgumentException("Un-parsable document ID " + id);
+
+ // we use the record archive position as the id number, and we then
+ // distribute the output files into a directory structure.
+ String posStr =
id.getAttributes().get(ARCInputHandler.RECORD_POSITION_ATTR);
+ // default value is 0, which means that all input records with no explicit
+ // position will end up in files called: 000/.../000/URL_as_file_name
+ long idNum = 0;
+ if(posStr != null && posStr.length() > 0) {
+ try {
+ idNum = Long.parseLong(posStr);
+ } catch (NumberFormatException e) {
+ logger.warn("Invalid record position value (not an integer number): " +
+ posStr);
+ }
}
- long idNum = Long.parseLong(idMatcher.group(1));
-
+
// build up the arguments for the format string by dividing the
// id number by the relevant divisors
Object[] formatArgs = new Object[divisors.length];
@@ -129,9 +140,12 @@
}
// write the numeric bit of the relative path
- new Formatter(pathBuilder).format(pattern, formatArgs);
+ Formatter formatter = new Formatter(pathBuilder);
+ formatter.format(pattern, formatArgs);
+ formatter.close();
- String remaining = idMatcher.group(2);
+ // the rest of the output file path is constructed from the record URL
+ String remaining = id.getIdText();
if(remaining != null && remaining.length() > 0) {
// append an underscore and the cleaned-up remaining part of the name
pathBuilder.append("_");
Modified: gcp/trunk/src/gate/cloud/io/arc/ARCInputHandler.java
===================================================================
--- gcp/trunk/src/gate/cloud/io/arc/ARCInputHandler.java 2013-03-22
02:43:36 UTC (rev 16611)
+++ gcp/trunk/src/gate/cloud/io/arc/ARCInputHandler.java 2013-03-22
10:27:20 UTC (rev 16612)
@@ -11,24 +11,38 @@
*/
package gate.cloud.io.arc;
-import static gate.cloud.io.IOConstants.*;
+import static gate.cloud.io.IOConstants.PARAM_ARC_FILE_LOCATION;
+import static gate.cloud.io.IOConstants.PARAM_BATCH_FILE_LOCATION;
+import static gate.cloud.io.IOConstants.PARAM_DEFAULT_ENCODING;
+import static gate.cloud.io.IOConstants.PARAM_MIME_TYPE;
+import gate.Document;
+import gate.Factory;
+import gate.FeatureMap;
+import gate.cloud.batch.DocumentID;
+import gate.cloud.io.DocumentData;
+import gate.cloud.io.InputHandler;
+import gate.cloud.util.ByteArrayURLStreamHandler;
+import gate.util.GateException;
import it.unimi.dsi.fastutil.longs.LongArrayList;
+
+import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
import java.net.URL;
+import java.net.URLConnection;
import java.text.ParseException;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.httpclient.Header;
import org.apache.commons.httpclient.util.DateParseException;
import org.apache.commons.httpclient.util.DateUtil;
+import org.apache.commons.io.IOUtils;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.log4j.Logger;
import org.archive.io.ArchiveRecord;
@@ -39,18 +53,26 @@
import org.archive.io.arc.ARCRecordMetaData;
import org.archive.util.ArchiveUtils;
-import gate.Document;
-import gate.Factory;
-import gate.FeatureMap;
-import gate.cloud.batch.DocumentID;
-import gate.cloud.batch.PooledDocumentProcessor;
-import gate.cloud.io.DocumentData;
-import gate.cloud.io.InputHandler;
-import gate.cloud.util.ByteArrayURLStreamHandler;
-import gate.util.GateException;
-
/**
* An input handler that reads from an ARC file.
+ * It assumes that the document IDs use the record URL as the id text
+ * (see {@link DocumentID#getIdText()}), and that the document ID attributes
+ * (see {@link DocumentID#getAttributes()}) include:
+ * <ul>
+ * <li>{@link #RECORD_OFFSET_ATTR}: the byte offset (inside the .arc file) for
+ * the required record.</li>
+ * <li>{@link #RECORD_LENGTH_ATTR}: the length (in bytes) for the required
+ * record.</li>
+ * <li>{@link #RECORD_POSITION_ATTR}: a numeric value for that
+ * is used as a sequence number. If the IDs are generated by an
+ * {@link ARCDocumentEnumerator}, the this attribute will contain the actual
+ * record position inside the .arc file.</li>
+ * <li>{@link #ARC_URL_ATTR} (optional): the full URL where the .arc file is
+ * available. This can be used to override the arc file configured on the
+ * actual input handler. For example, when a batch uses records from multiple
+ * .arc file as input, the arc file cannot be configured on the input handler,
+ * so each document ID must include one instead.
+ * </ul>
*/
public class ARCInputHandler implements InputHandler {
@@ -65,18 +87,27 @@
/**
* Name for an attribute used when generating {@link DocumentID} values for
* the {@link ARCInputHandler}. This attribute stores (a String
- * representation) of the Long byte offset for the input .record
+ * representation) of the Long byte offset for the input record.
+ *
+ * Value: {@value}
*/
static final String RECORD_OFFSET_ATTR = "recordOffset";
/**
* Name for an attribute used when generating {@link DocumentID} values for
* the {@link ARCInputHandler}. This attribute stores (a String
- * representation) of the Long byte length for the input .record
+ * representation) of the Long byte length for the input record.
*/
static final String RECORD_LENGTH_ATTR = "recordLength";
+
+ /**
+ * Name for an attribute used when generating {@link DocumentID} values for
+ * the {@link ARCInputHandler}. This attribute stores (a String
+ * representation) of the integer position inside the archive for the input
+ * record.
+ */
+ static final String RECORD_POSITION_ATTR = "recordPosition";
-
/**
* Name for an attribute used when generating {@link DocumentID} values for
* the {@link ARCInputHandler}. This attribute stores the URL where the input
@@ -115,22 +146,6 @@
protected String mimeType;
/**
- * Pool of archive readers.
- */
-// protected BlockingQueue<ARCReader> readers;
-
- /**
- * The index from archive record number to offset.
- */
- protected long[] index;
-
- /**
- * Regular expression pattern matching any sequence of digits
- * at the start of a string.
- */
- protected static final Pattern STARTING_DIGITS_PATTERN =
Pattern.compile("^(\\d+)");
-
- /**
* Regular expression pattern matching the "charset" from an HTTP
* Content-type header.
*/
@@ -140,29 +155,28 @@
GateException {
// ARC file
String arcFileStr = configData.get(PARAM_ARC_FILE_LOCATION);
- if(arcFileStr == null || arcFileStr.trim().length() == 0) {
- throw new IllegalArgumentException(
- "No value was provided for the required parameter \""
- + PARAM_ARC_FILE_LOCATION + "\"!");
+ if(arcFileStr != null) {
+ String batchFileStr = configData.get(PARAM_BATCH_FILE_LOCATION);
+ if(batchFileStr != null) {
+ batchDir = new File(batchFileStr).getParentFile();
+ }
+ arcFile = new File(arcFileStr);
+ if(!arcFile.isAbsolute()) {
+ arcFile = new File(batchDir, arcFileStr);
+ }
+ if(!arcFile.exists()) {
+ throw new IllegalArgumentException("File \"" + arcFile
+ + "\", provided as value for required parameter \""
+ + PARAM_ARC_FILE_LOCATION + "\", does not exist!");
+ }
+ if(!arcFile.isFile()) {
+ throw new IllegalArgumentException("File \"" + arcFile
+ + "\", provided as value for required parameter \""
+ + PARAM_ARC_FILE_LOCATION + "\", is not a file!");
+ }
+ } else {
+ arcFile = null;
}
- String batchFileStr = configData.get(PARAM_BATCH_FILE_LOCATION);
- if(batchFileStr != null) {
- batchDir = new File(batchFileStr).getParentFile();
- }
- arcFile = new File(arcFileStr);
- if(!arcFile.isAbsolute()) {
- arcFile = new File(batchDir, arcFileStr);
- }
- if(!arcFile.exists()) {
- throw new IllegalArgumentException("File \"" + arcFile
- + "\", provided as value for required parameter \""
- + PARAM_ARC_FILE_LOCATION + "\", does not exist!");
- }
- if(!arcFile.isFile()) {
- throw new IllegalArgumentException("File \"" + arcFile
- + "\", provided as value for required parameter \""
- + PARAM_ARC_FILE_LOCATION + "\", is not a file!");
- }
defaultEncoding = configData.get(PARAM_DEFAULT_ENCODING);
if(defaultEncoding == null) {
@@ -172,87 +186,68 @@
mimeType = configData.get(PARAM_MIME_TYPE);
}
- public void init() throws IOException, GateException {
- // build index
- logger.info("Building index for ARC file " + arcFile.getAbsolutePath()
- + " - this may take some time...");
- int recordNum = 0;
- ARCReader reader = ARCReaderFactory.get(arcFile);
- LongArrayList indexList = new LongArrayList();
- try {
- Iterator<ArchiveRecord> iter = reader.iterator();
- while(iter.hasNext()) {
- ARCRecord record = (ARCRecord)iter.next();
- indexList.add(record.getMetaData().getOffset());
- if(recordNum++ % 1000 == 0) {
- logger.debug("Processed " + (recordNum - 1) + " records");
- }
- }
+ public void init() throws IOException, GateException { }
+
+ public DocumentData getInputDocument(DocumentID id) throws IOException,
GateException {
+ if(id.getAttributes() == null) {
+ throw new IllegalArgumentException(
+ "Document IDs within an ARC must include \"" + RECORD_OFFSET_ATTR +
+ "\" and \"" + RECORD_LENGTH_ATTR + "\" attributes; \"" +
+ id.toString() + "\" does not.");
}
- finally {
- reader.close();
+ String arcUrlStr = id.getAttributes().get(ARC_URL_ATTR);
+ if(arcFile == null && (arcUrlStr == null || arcUrlStr.length() == 0)) {
+ throw new IllegalArgumentException(
+ "No .arc file was configured for this input handler, so the " +
+ "document IDs must provide a \"" + ARC_URL_ATTR +
+ "\" attribute; \"" + id.toString() + "\" does not.");
}
- // turn the list into an array
- indexList.trim();
- index = indexList.elements();
- logger.info("Finished indexing ARC file " + arcFile.getAbsolutePath()
- + ", " + index.length + " records total.");
-
-// readers = new LinkedBlockingQueue<ARCReader>();
- }
-
- public DocumentData getInputDocument(DocumentID id) throws IOException,
GateException {
- ARCReader reader = ARCReaderFactory.get(arcFile);
- reader.setParseHttpHeaders(true);
-
- try {
-// Matcher idMatcher = STARTING_DIGITS_PATTERN.matcher(id.getIdText());
-// if(!idMatcher.find()) {
-// throw new GateException("Document IDs within an ARC must start " +
-// "with one or more digits, \"" + id + "\" does not.");
-// }
-// int docIndex = Integer.parseInt(idMatcher.group(1));
-// if(docIndex >= index.length) {
-// throw new GateException("Input document at index " + docIndex + "
requested, " +
-// "but this ARC file only contains " + index.length + "
entries.");
-// }
-//
-//
-// // get the right ARC record
-// ARCRecord record = (ARCRecord)reader.get(index[docIndex]);
-
- if(id.getAttributes() == null) {
- throw new IllegalArgumentException(
+ String offsetStr = id.getAttributes().get(RECORD_OFFSET_ATTR);
+ String lengthStr = id.getAttributes().get(RECORD_LENGTH_ATTR);
+ if(offsetStr == null || offsetStr.length() == 0 ||
+ lengthStr == null || lengthStr.length() == 0) {
+ throw new IllegalArgumentException(
"Document IDs within an ARC must include \"" + RECORD_OFFSET_ATTR +
"\" and \"" + RECORD_LENGTH_ATTR + "\" attributes; \"" +
- id.toString() + "\" does not.");
+ id.toString() + "\" does not.");
+ }
+
+ long offset = -1;
+ try{
+ offset = Long.parseLong(offsetStr);
+ }catch (NumberFormatException nfe) {
+ throw new IllegalArgumentException(
+ "Could not parse offset \"" + offsetStr + "\" as a long value.");
+ }
+ long length = -1;
+ try{
+ length = Long.parseLong(lengthStr);
+ }catch (NumberFormatException nfe) {
+ throw new IllegalArgumentException(
+ "Could not parse length \"" + lengthStr + "\" as a long value.");
+ }
+ // read the arc record
+ ARCRecord record = null;
+ ARCReader reader = null;
+ try{
+ if(arcUrlStr != null) {
+ URL arcUrl = new URL(arcUrlStr);
+ URLConnection conn = arcUrl.openConnection();
+ conn.setRequestProperty("Range", "bytes=" + Long.toString(offset) +
+ "-" + Long.toString(offset + length - 1));
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ InputStream is = conn.getInputStream();
+ IOUtils.copy(is, baos);
+ is.close();
+ record = new ARCRecord(new ByteArrayInputStream(baos.toByteArray()),
+ arcUrlStr, 0, false, false, true);
+ } else {
+ // no custom URL, so we can use the default arc file
+ reader = ARCReaderFactory.get(arcFile);
+ reader.setParseHttpHeaders(true);
+ record = (ARCRecord)reader.get(offset);
}
- String offsetStr = id.getAttributes().get(RECORD_OFFSET_ATTR);
- String lengthStr = id.getAttributes().get(RECORD_LENGTH_ATTR);
- if(offsetStr == null || offsetStr.length() == 0 ||
- lengthStr == null || lengthStr.length() == 0) {
- throw new IllegalArgumentException(
- "Document IDs within an ARC must include \"" + RECORD_OFFSET_ATTR +
- "\" and \"" + RECORD_LENGTH_ATTR + "\" attributes; \"" +
- id.toString() + "\" does not.");
- }
-
- long offset = -1;
- try{
- offset = Long.parseLong(offsetStr);
- }catch (NumberFormatException nfe) {
- throw new IllegalArgumentException(
- "Could not parse offset \"" + offsetStr + "\" as a long value.");
- }
- long length = -1;
- try{
- length = Long.parseLong(lengthStr);
- }catch (NumberFormatException nfe) {
- throw new IllegalArgumentException(
- "Could not parse length \"" + lengthStr + "\" as a long value.");
- }
- ARCRecord record = (ARCRecord)reader.get(offset);
ARCRecordMetaData header = record.getMetaData();
// extract the content
@@ -298,7 +293,19 @@
docFeatures.put("redirect_to", redirect.toString());
}
docFeatures.put("archive_name", arcFile.getName());
-// docFeatures.put("archive_position", idMatcher.group(1));
+ // docFeatures.put("archive_position", idMatcher.group(1));
+ String posStr = id.getAttributes().get(RECORD_POSITION_ATTR);
+ if(posStr != null) {
+ try {
+ long pos = Long.parseLong(posStr);
+ docFeatures.put("archive_position", posStr);
+ } catch(NumberFormatException e) {
+ // log and ignore
+ logger.warn("Invalid record position value (not an integer number):
" +
+ posStr);
+ }
+ }
+
Date documentDate = getDate(header, record);
if(documentDate != null) {
docFeatures.put("retrievedAt", documentDate);
@@ -328,7 +335,7 @@
}
// Do the same for the HTTP headers
Header[] httpHeader = record.getHttpHeaders();
-
+
if(httpHeader != null) {
for(Header h : httpHeader) {
headerKey = h.getName();
@@ -344,11 +351,12 @@
return docData;
} finally {
try {
- reader.close();
+ if(reader != null) reader.close();
} catch(IOException e) {
// ignore
logger.error("Error while closing ARC reader.", e);
}
+ if(record != null) record.close();
}
}
Modified: gcp/trunk/src/gate/cloud/io/file/SimpleNamingStrategy.java
===================================================================
--- gcp/trunk/src/gate/cloud/io/file/SimpleNamingStrategy.java 2013-03-22
02:43:36 UTC (rev 16611)
+++ gcp/trunk/src/gate/cloud/io/file/SimpleNamingStrategy.java 2013-03-22
10:27:20 UTC (rev 16612)
@@ -100,7 +100,7 @@
public File toFile(DocumentID id) throws IOException {
try {
- String path = relativePathFor(id.getIdText());
+ String path = relativePathFor(id);
if(fileExtension != null) {
path += fileExtension;
}
@@ -125,8 +125,8 @@
* @param id the document ID
* @return a path suitable for use as a relative URI.
*/
- protected String relativePathFor(String id) {
- return id;
+ protected String relativePathFor(DocumentID id) {
+ return id.getIdText();
}
public String toString() {
This was sent by the SourceForge.net collaborative development platform, the
world's largest Open Source development site.
------------------------------------------------------------------------------
Everyone hates slow websites. So do we.
Make your web apps faster with AppDynamics
Download AppDynamics Lite for free today:
http://p.sf.net/sfu/appdyn_d2d_mar
_______________________________________________
GATE-cvs mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/gate-cvs
