D-Net

package eu.dnetlib.iis.importer.acm;

import java.io.InputStream;

import java.security.InvalidParameterException;

import java.util.Collections;

import java.util.HashMap;

import java.util.Map;

import javax.xml.parsers.SAXParser;

import javax.xml.parsers.SAXParserFactory;

import org.apache.avro.file.DataFileWriter;

import org.apache.hadoop.fs.FileStatus;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import eu.dnetlib.iis.citationmatching.schemas.DocumentMetadata;

import eu.dnetlib.iis.core.java.HadoopContext;

import eu.dnetlib.iis.core.java.PortBindings;

import eu.dnetlib.iis.core.java.Process;

import eu.dnetlib.iis.core.java.io.DataStore;

import eu.dnetlib.iis.core.java.io.FileSystemPath;

import eu.dnetlib.iis.core.java.porttype.AvroPortType;

import eu.dnetlib.iis.core.java.porttype.PortType;

import eu.dnetlib.iis.importer.dataset.DataFileRecordReceiver;

/**

 * Process module importing ACM records from xml dump

 * and writing output to avro datastore.

 * @author mhorst

 *

 */

public class AcmXmlImporter implements Process {

	private static final String PORT_OUT_DOCUMENT_METADATA = "document_metadata";

	public static final String PARAM_ACM_XML_DUMP_PATH = "import.acm.xmldump.path";

	private static final Map<String, PortType> outputPorts = new HashMap<String, PortType>();

	{

		outputPorts.put(PORT_OUT_DOCUMENT_METADATA, 

				new AvroPortType(DocumentMetadata.SCHEMA$));

	}

	@Override

	public Map<String, PortType> getInputPorts() {

		return Collections.emptyMap();

	}

	@Override

	public Map<String, PortType> getOutputPorts() {

		return outputPorts;

	}

	@Override

	public void run(PortBindings portBindings, HadoopContext context,

			Map<String, String> parameters) throws Exception {

		FileSystem fs = FileSystem.get(context.getConfiguration());

		if (parameters.containsKey(PARAM_ACM_XML_DUMP_PATH)) {

			DataFileWriter<DocumentMetadata> datasetRefWriter = null;

			try {

				datasetRefWriter = DataStore.create(

						new FileSystemPath(fs, portBindings.getOutput().get(PORT_OUT_DOCUMENT_METADATA)), 

						DocumentMetadata.SCHEMA$);

				processNode(fs, 

						new Path(parameters.get(PARAM_ACM_XML_DUMP_PATH)),

						datasetRefWriter);

			} finally {

				if (datasetRefWriter!=null) {

					datasetRefWriter.close();	

				}	

			}		

		} else {

			throw new InvalidParameterException("required parameter '" + 

					PARAM_ACM_XML_DUMP_PATH + "' is missing!");

		}

	}

	protected void processNode(FileSystem fs, Path currentPath,

			DataFileWriter<DocumentMetadata> datasetRefWriter) throws Exception {

		if (fs.isDirectory(currentPath)) {

			for (FileStatus fileStatus : fs.listStatus(currentPath)) {

				processNode(fs, fileStatus.getPath(), 

						datasetRefWriter);

			}

		} else {

			InputStream inputStream = null;

			SAXParser saxParser = null;

			try {

				saxParser = SAXParserFactory.newInstance().newSAXParser();

				saxParser.parse(inputStream = fs.open(

						currentPath),

						new AcmDumpXmlHandler( 

								new DataFileRecordReceiver<DocumentMetadata>(datasetRefWriter)));	

			} finally {

				if (inputStream!=null) {

					inputStream.close();

				}	

			}

		}

	}

}

package eu.dnetlib.iis.importer.acm;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Collections;

import java.util.List;

import java.util.Stack;

import org.apache.log4j.Logger;

import org.xml.sax.Attributes;

import org.xml.sax.SAXException;

import org.xml.sax.helpers.DefaultHandler;

import eu.dnetlib.iis.citationmatching.schemas.BasicMetadata;

import eu.dnetlib.iis.citationmatching.schemas.DocumentMetadata;

import eu.dnetlib.iis.citationmatching.schemas.ReferenceMetadata;

import eu.dnetlib.iis.importer.dataset.RecordReceiver;

/**

 * ACM XML dump SAX handler.

 * Notice: writer is not being closed by handler.

 * Created outside, let it be closed outside as well.

 * @author mhorst

 *

 */

public class AcmDumpXmlHandler extends DefaultHandler {

	private static final String ELEM_ARTICLE_REC = "article_rec";

	private static final String ELEM_ARTICLE_ID = "article_id";

	private static final String ELEM_TITLE = "title";

	private static final String ELEM_REFERENCES = "references";

	private static final String ELEM_REF = "ref";

	private static final String ELEM_REF_SEQ_NO = "ref_seq_no";

	private static final String ELEM_REF_TEXT = "ref_text";

	private final Logger log = Logger.getLogger(this.getClass());

	private Stack<String> parents;

	private StringBuilder currentValue = new StringBuilder();

	private String articleId = null;

	private String title = null;

	private List<ReferenceMetadata> references = null;	

	private Integer refSeqNo = null;

	private String refText = null;

	private int counter = 0;

	private final RecordReceiver<DocumentMetadata> receiver;

	/**

	 * Default constructor.

	 * @param receiver

	 */

	public AcmDumpXmlHandler(RecordReceiver<DocumentMetadata> receiver) {

		super();

		this.receiver = receiver;

	}

	@Override

	public void startDocument() throws SAXException {

		this.parents = new Stack<String>();

		this.counter = 0;

		clearAllFields();

	}

	@Override

	public void startElement(String uri, String localName, String qName,

			Attributes attributes) throws SAXException {

		if (isWithinElement(qName, ELEM_ARTICLE_ID, ELEM_ARTICLE_REC)) {

			this.currentValue = new StringBuilder();

		} else if (isWithinElement(qName, ELEM_TITLE, ELEM_ARTICLE_REC)) {

			this.currentValue = new StringBuilder();

		} else if (isWithinElement(qName, ELEM_REFERENCES, ELEM_ARTICLE_REC)) {

			this.currentValue = new StringBuilder();

			this.references = new ArrayList<ReferenceMetadata>();

		} else if (isWithinElement(qName, ELEM_REF_SEQ_NO, ELEM_REF)) {

			this.currentValue = new StringBuilder();

		} else if (isWithinElement(qName, ELEM_REF_TEXT, ELEM_REF)) {

			this.currentValue = new StringBuilder();

		}

		this.parents.push(qName);

	}

	@Override

	public void endElement(String uri, String localName, String qName)

			throws SAXException {

		this.parents.pop();

		if (isWithinElement(qName, ELEM_ARTICLE_ID, ELEM_ARTICLE_REC)) {

			this.articleId = this.currentValue.toString().trim();

		} else if (isWithinElement(qName, ELEM_TITLE, ELEM_ARTICLE_REC)) {

			this.title = this.currentValue.toString().trim();

		} else if (isWithinElement(qName, ELEM_REF_SEQ_NO, ELEM_REF)) {

			this.refSeqNo = Integer.parseInt(this.currentValue.toString().trim());

		} else if (isWithinElement(qName, ELEM_REF_TEXT, ELEM_REF)) {

			this.refText = this.currentValue.toString().trim();

		} else if (isWithinElement(qName, ELEM_REF, ELEM_REFERENCES)) {

			ReferenceMetadata.Builder refMetaBuilder = ReferenceMetadata.newBuilder();

			BasicMetadata.Builder basicMetaBuilder = BasicMetadata.newBuilder();

			basicMetaBuilder.setAuthors(Collections.<CharSequence>emptyList());

			refMetaBuilder.setBasicMetadata(basicMetaBuilder.build());

			refMetaBuilder.setPosition(this.refSeqNo);

			refMetaBuilder.setRawText(this.refText);

			this.references.add(refMetaBuilder.build());

		} else if (isWithinElement(qName, ELEM_ARTICLE_REC, null)) {

//			writing whole record

			if (this.articleId!=null && !this.articleId.isEmpty()) {

				try {

					DocumentMetadata.Builder docMetaBuilder = DocumentMetadata.newBuilder();

					docMetaBuilder.setId(this.articleId);

					BasicMetadata.Builder basicMetaBuilder = BasicMetadata.newBuilder();

					basicMetaBuilder.setAuthors(Collections.<CharSequence>emptyList());

					if (this.title!=null && !this.title.isEmpty()) {

						basicMetaBuilder.setTitle(this.title);	

					}

					docMetaBuilder.setBasicMetadata(basicMetaBuilder.build());

					docMetaBuilder.setReferences(this.references);

					receiver.receive(docMetaBuilder.build());

					counter++;

					if (counter%10000==0) {

						log.debug("current progress: " + counter);

					}

				} catch (IOException e) {

					throw new SAXException(e);

				}

			} else {

				log.warn("omitting record with null/empty article id and title: " + this.title);

			}

			clearAllFields();

		}

//		resetting current value;

		this.currentValue = null;

	}

	private void clearAllFields() {

		this.articleId = null;

		this.title = null;

		this.references = null;		

		this.refSeqNo = null;

		this.refText = null;

	}

	boolean isWithinElement(String qName,

			String expectedElement, String expectedParent) {

		return qName.equals(expectedElement) && 

				(expectedParent==null || !this.parents.isEmpty() && expectedParent.equals(this.parents.peek()));

	}

	@Override

	public void endDocument() throws SAXException {

		parents.clear();

		parents = null;

		log.debug("total number of processed records: " + counter);

	}

	@Override

	public void characters(char[] ch, int start, int length)

			throws SAXException {

		if (this.currentValue!=null) {

			this.currentValue.append(ch, start, length);

		}

	}

}

<workflow-app xmlns="uri:oozie:workflow:0.4" name="importer_acm">

	<parameters>

		<property>

			<name>input_hdfs_location</name>

			<description>input ACM XML dump HDFS location</description>

		</property>

		<property>

			<name>output</name>

			<description>document metadata output directory</description>

		</property>

	</parameters>

	<start to="acm-importer" />

	<action name="acm-importer">

		<java>

			<job-tracker>${jobTracker}</job-tracker>

			<name-node>${nameNode}</name-node>

			<!-- The data generated by this node is deleted in this section -->

			<prepare>

				<delete path="${nameNode}${workingDir}/import_dataset" />

				<delete path="${nameNode}${output}" />

				<mkdir path="${nameNode}${workingDir}/import_dataset" />

			</prepare>

			<configuration>

				<property>

					<name>mapred.job.queue.name</name>

					<value>${queueName}</value>

				</property>

			</configuration>

			<main-class>eu.dnetlib.iis.core.java.ProcessWrapper</main-class>

			<arg>eu.dnetlib.iis.importer.acm.AcmXmlImporter</arg>

			<arg>-SworkingDir=${workingDir}</arg>

			<arg>-Pimport.acm.xmldump.path=${input_hdfs_location}</arg>

			<arg>-Odocument_metadata=${output}</arg>

		</java>

		<ok to="end" />

		<error to="fail" />

	</action>

	<kill name="fail">

		<message>Unfortunately, the process failed -- error message:

			[${wf:errorMessage(wf:lastErrorNode())}]</message>

	</kill>

	<end name="end" />

</workflow-app>

#input_hdfs_location=/tmp/acm_citations/sample

input_hdfs_location=/tmp/acm_citations/full

output=${workingDir}/out