D-Net

define avro_load_document_content

org.apache.pig.piggybank.storage.avro.AvroStorage(

'input_schema_class', '$schema_document_content');

define avro_load_document_text

org.apache.pig.piggybank.storage.avro.AvroStorage(

'input_schema_class', '$schema_document_text');

define avro_store_document_content

org.apache.pig.piggybank.storage.avro.AvroStorage(

'index', '0',

'output_schema_class', '$schema_document_content');

define avro_store_document_text

org.apache.pig.piggybank.storage.avro.AvroStorage(

'index', '1',

'output_schema_class', '$schema_document_text');

documentContent = load '$input_document_content' using avro_load_document_content;

documentText = load '$input_document_text' using avro_load_document_text;

documentTextId = foreach documentText generate id;

cachedDocumentIdDistinct = distinct documentTextId;

joinedDocumentContent = join documentContent by id left, cachedDocumentIdDistinct by id;

joinedFilteredDocumentContent = filter joinedDocumentContent by cachedDocumentIdDistinct::id is null;

documentContentFiltered = foreach joinedFilteredDocumentContent generate documentContent::id as id, documentContent::url as url, documentContent::mimeType as mimeType, documentContent::contentChecksum as contentChecksum;

documentContentId = foreach documentContent generate id;

documentContentIdDistinct = distinct documentContentId;

joinedDocumentText = join documentText by id, documentContentIdDistinct by id;

documentTextFiltered = foreach joinedDocumentText generate documentText::id as id, documentText::text as text;

store documentContentFiltered into '$output_document_content' using avro_store_document_content;

store documentTextFiltered into '$output_document_text' using avro_store_document_text;

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

	<parameters>

		<property>

			<name>input_document_content</name>

			<description>document content input</description>

		</property>

		<property>

			<name>input_document_text</name>

			<description>document text input</description>

		</property>

		<property>

			<name>output_document_content</name>

			<description>document content output: all contents which were not processed so far, based in input_document_meta inspection</description>

		</property>

		<property>

			<name>output_document_text</name>

			<description>document text ouput: all plaintext records which were already processed, found in input_document_text</description>

		</property>

	</parameters>

    <start to="transformer"/>

    <action name="transformer">

        <pig>

            <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}/transformer" />

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

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

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

			</prepare>

            <configuration>

                <property>

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

                    <value>${queueName}</value>

                </property>

                <property>

                    <name>mapred.map.child.java.opts</name>

                    <value>-Xmx4g</value>

                </property>

                <property>

                    <name>mapred.reduce.child.java.opts</name>

                    <value>-Xmx4g</value>

                </property>

            </configuration>

            <!-- Path to PIG script the workflow executes. -->

            <script>lib/scripts/transformer.pig</script>

            <!-- The working directory of the workflow node. -->

            <param>workingDir=${workingDir}/transformer/working_dir</param>

            <param>schema_document_content=eu.dnetlib.iis.importer.auxiliary.schemas.DocumentContentUrl</param>

            <param>schema_document_text=eu.dnetlib.iis.metadataextraction.schemas.DocumentText</param>

            <param>input_document_content=${input_document_content}</param>

            <param>input_document_text=${input_document_text}</param>

            <param>output_document_content=${output_document_content}</param>

            <param>output_document_text=${output_document_text}</param>

        </pig>

        <ok to="end"/>

        <error to="fail"/>

    </action>

    <kill name="fail">

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

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

    </kill>

    <end name="end"/>

</workflow-app>

input_document_content=/share/import/doc_content_url/europePMC/2014-06-19

input_document_text=/cache/plaintext/europePMC

output_document_content=${workingDir}/tobeprocessed_content

output_document_text=${workingDir}/tobereturned_plaintext