/ - Diff - D-Net - D-Net project tracking tool

                 <version>1.8.5</version>
                 <scope>test</scope>
             </dependency>
             <dependency>
                 <groupId>eu.dnetlib</groupId>
                 <artifactId>dnet-msro-service</artifactId>
                 <version>5.0.0-SNAPSHOT</version>
             </dependency>
         </dependencies>
     </project>

     #dnet.logger.mongo.host=localhost
     dnet.logger.mongo.port=27017
     dnet.modular.ui.authorization.mongo.host=playground-t.dnet.d4science.org
     services.mdstore.mongodb.host=playground-t.dnet.d4science.org
     #services.mdstore.mongodb.host=playground-t.dnet.d4science.org
     #dnet.modular.ui.authorization.mongo.host=localhost
     #services.mdstore.mongodb.host=localhost
     services.dli.resolver.store.DatabaseName=resolvedStore
     services.mdstore.mongodb.host=localhost
     services.dli.resolver.store.DatabaseName=dliResolvedStore
     services.dli.resolver.crossRef.dump=ES

     package eu.dnetlib.msro.workflows.nodes.transform;
     import eu.dnetlib.enabling.locators.UniqueServiceLocator;
     import eu.dnetlib.msro.logging.DnetLogger;
     import eu.dnetlib.msro.workflows.graph.Arc;
     import eu.dnetlib.msro.workflows.nodes.DateProcessUtils;
     import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;
     import eu.dnetlib.msro.workflows.procs.Env;
     import eu.dnetlib.msro.workflows.procs.ProcessAware;
     import eu.dnetlib.msro.workflows.procs.WorkflowProcess;
     import eu.dnetlib.msro.workflows.util.WorkflowsConstants;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.springframework.beans.factory.annotation.Autowired;
     import java.util.HashMap;
     import java.util.Map;
     public class IncrementalTransformationJobNode extends SimpleJobNode implements ProcessAware {
         enum TransformationType {
             INCREMENTAL,
             REFRESH
+        }
         private static final Log log = LogFactory.getLog(IncrementalTransformationJobNode.class);
         private String transformationType;
         private WorkflowProcess process;
         @Autowired
         private DnetLogger dnetLogger;
         @Autowired
         private UniqueServiceLocator locator;
         @Override
         protected String execute(Env env) throws Exception {
             if (TransformationType.INCREMENTAL.toString().equalsIgnoreCase(getTransformationType())) {
                 final String endDate = DateProcessUtils.getEndDate(process, dnetLogger);
                 if (endDate == null )
+                {
                     env.setAttribute("collectionMode", TransformationType.REFRESH.toString());
+                }
                 else {
                     env.setAttribute("collectionMode", TransformationType.INCREMENTAL.toString());
                     env.setAttribute("incrementalDateFrom", endDate);
+                }
+            }
             return Arc.DEFAULT_ARC;
+        }
         public void setTransformationType(String transformationType) {
             this.transformationType = transformationType;
+        }
         public String getTransformationType() {
             return transformationType;
+        }
         @Override
         public void setProcess(WorkflowProcess process) {
             this.process = process;
+        }
+    }

     				<PROPERTY key="mapreduce.map.speculative" value="false"/>
     				<PROPERTY key="mapreduce.reduce.speculative" value="false"/>
     				<PROPERTY key="mapred.reduce.tasks" value="1"/>
     				<PROPERTY key="mapred.reduce.tasks" value="100"/>
     				<!-- <PROPERTY key="user.name" value="dnet" /> -->
     				<!--  	Uncomment to override the default lib path -->

     		final ResultSet<?> rs = token.getEnv().getAttribute(getEprParam(), ResultSet.class);
     		token.setProgressProvider(new ResultsetProgressProvider(rs, this.resultSetClient));
     		final String collectionMode = token.getEnv().getAttribute("collectionMode", String.class);
     		job.setAction("FEED");
     		job.getParameters().put("epr", rs.toJson());
     		job.getParameters().put("storingType", getStoringType());
     		job.getParameters().put("storingType",collectionMode!=null?collectionMode:getStoringType());
     		job.getParameters().put("mdId", getMdId());
+    	}

+        }
         @Test
         public void testENATransform() throws Exception {
             final InputStream xsltAsStream =
                     getClass().getResourceAsStream("/eu/dnetlib/dli/transform/ena.xslt");
             final TransformerFactory factory = TransformerFactory.newInstance();
             factory.newTransformer();
             final StreamSource xsltSource = new StreamSource(xsltAsStream);
             final InputStream recordStream = this.getClass().getResourceAsStream("/eu/dnetlib/dli/transform/input_ena.xml");
             final Transformer transformer = factory.newTransformer(xsltSource);
             transformer.setOutputProperty(OutputKeys.INDENT, "yes");
             final StringWriter output = new StringWriter();
             transformer.transform(new StreamSource(recordStream), new StreamResult(output));
             System.out.println(output.toString());
+        }
         @Test
         public void testIEEETransform() throws Exception {
             final InputStream xsltAsStream =

     <?xml version="1.0" encoding="UTF-8"?>
     <oai:record xmlns:oai="http://www.openarchives.org/OAI/2.0/"
                 xmlns:dri="http://www.driver-repository.eu/namespace/dri"
                 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
         <oai:header>
             <dri:objIdentifier xmlns:oaf="http://namespace.dnet.eu/oaf"
                                xmlns:datacite="http://datacite.org/schema/kernel-3">r3d100010527::00000648e410cb638823564cf5fdc885</dri:objIdentifier>
             <dri:recordIdentifier xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">http://www.ebi.ac.uk/ena/data/search/?query=11118631&amp;result=sequence_release::url</dri:recordIdentifier>
             <dri:dateOfCollection xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">2017-09-18T14:07:46.84+02:00</dri:dateOfCollection>
             <dri:repositoryId xmlns:oaf="http://namespace.dnet.eu/oaf"
                               xmlns:datacite="http://datacite.org/schema/kernel-3">dli_________::r3d100010527</dri:repositoryId>
             <dri:datasourceprefix xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">r3d100010527</dri:datasourceprefix>
         </oai:header>
         <metadata xmlns:oaf="http://namespace.dnet.eu/oaf"
                   xmlns:datacite="http://datacite.org/schema/kernel-3"
                   xmlns:dc="http://purl.org/dc/elements/1.1/">
             <resource xmlns="http://datacite.org/schema/kernel-3"
                       xsi:schemaLocation="http://datacite.org/schema/kernel-3 http://schema.datacite.org/meta/kernel-3/metadata.xsd">
                 <identifier xmlns="" identifierType="url">http://www.ebi.ac.uk/ena/data/search/?query=11118631&amp;result=sequence_release</identifier>
                 <titles xmlns="">
                     <title>Nucleotide sequences (Release) (1/2) of "Molecular analysis of a t(7;14)(q35;q32) chromosome translocation in a T cell leukemia of a patient with ataxia telangiectasia." </title>
                 </titles>
                 <dates xmlns="">
                     <date dateType="Collected">2000-04-21</date>
                 </dates>
                 <descriptions xmlns="">
                     <description>Catharanthus roseus 2C-methyl-D-erythritol 2,4-cyclodiphosphate synthase (MECS) mRNA, complete cds.</description>
                 </descriptions>
                 <subjects xmlns="">
                     <subject subjectScheme="organism">Catharanthus roseus</subject>
                     <subject subjectScheme="codon_start">1</subject>
                     <subject subjectScheme="gene">MECS</subject>
                     <subject subjectScheme="product">2C-methyl-D-erythritol 2,4-cyclodiphosphate synthase</subject>
                     <subject subjectScheme="note">YgbB protein</subject>
                     <subject subjectScheme="protein_id">AAF65155.1</subject>
                     <subject subjectScheme="translation">MAMATSFYCSTAIPSKKTNQNRENFLCSPVGGSKTTPSYIRLSTRQSRTLSLVVSAAASGAAVEAEPKFAAVTPSKILSF
                         RVGHGFDLHRLEPGYPLIIGGINIPHDRGCEAHSDGDVLLHCVVDAILGALGLPDIGQIFPDTDPKWKGAPSSVFIKEAV
                         RLMDEAGYELGNLDATLILQRPKVSPHKEAIRQNLCQLLGADPCVVNLKAKTHEKVDSLGENRSIAAHTVVLLMRK</subject>
                 </subjects>
                 <resourceType xmlns="" resourceTypeGeneral="dataset">dataset</resourceType>
                 <oaf:relatedIdentifier xmlns="" entityType="publication" inverseRelationType="isRelatedTo"
                                        relatedIdentifierType="dnet"
                                        relationType="isRelatedTo">r3d100010527::f89cc26f592efddee8437a6a73a6ab5a</oaf:relatedIdentifier>
             </resource>
         </metadata>
         <oai:about>
             <provenance xmlns="http://www.openarchives.org/OAI/2.0/provenance"
                         xsi:schemaLocation="http://www.openarchives.org/OAI/2.0/provenance http://www.openarchives.org/OAI/2.0/provenance.xsd">
                 <originDescription xmlns="" altered="true" harvestDate="2018-01-24T16:52:33.227+01:00">
                     <baseURL>%2Fsrv%2Fmongo%2Fdli_import%2Fena.gz</baseURL>
                     <identifier/>
                     <datestamp/>
                     <metadataNamespace/>
                 </originDescription>
             </provenance>
         </oai:about>
     </oai:record>

     		class="eu.dnetlib.msro.workflows.nodes.transform.TransformJobNode"
     		scope="prototype"/>
     	<bean id="wfNodeIncrementalTransformation"
     		  class="eu.dnetlib.msro.workflows.nodes.transform.IncrementalTransformationJobNode"
     		  scope="prototype"/>
     	<bean id="wfNodeApplyXslt"
     		class="eu.dnetlib.msro.workflows.nodes.transform.ApplyXsltJobNode"
     		scope="prototype"/>

     <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
                     xmlns:dri="http://www.driver-repository.eu/namespace/dri"
                     xmlns:dc="http://purl.org/dc/elements/1.1/"
                     xmlns:oaf="http://namespace.dnet.eu/oaf" version="2.0" exclude-result-prefixes="xsl">
         <xsl:template match="/">
             <xsl:apply-templates mode="dli"></xsl:apply-templates>
         </xsl:template>
         <xsl:template match="@*|node()" mode="dli">
             <xsl:copy>
                 <xsl:apply-templates select="@*|node()" mode="dli"/>
             </xsl:copy>
         </xsl:template>
         <xsl:template match="*[local-name()='about']" mode="dli">
             <oaf:about>
                 <oaf:datainfo>
                     <oaf:completionStatus>complete</oaf:completionStatus>
                     <xsl:choose>
                         <xsl:when test="//dc:type='publication'">
                             <oaf:collectedFrom completionStatus="incomplete" id="r3d100010527" name="European Nucleotide Archive" />
                             <oaf:resolvedFrom completionStatus="complete" id="dli_________::pubmed" name="PubMed"/>
                             <oaf:provisionMode>resolved</oaf:provisionMode>
                         </xsl:when>
                         <xsl:otherwise>
                             <oaf:collectedFrom completionStatus="complete" id="r3d100010527" name="European Nucleotide Archive" />
                             <oaf:provisionMode>collected</oaf:provisionMode>
                         </xsl:otherwise>
                     </xsl:choose>
                 </oaf:datainfo>
             </oaf:about>
         </xsl:template>
     </xsl:stylesheet>

     import org.bson.Document;
     import org.bson.conversions.Bson;
     import org.springframework.beans.factory.annotation.Autowired;
     import org.springframework.beans.factory.annotation.Required;
     // TODO: Auto-generated Javadoc
-...
     	@Autowired
     	private CrossRefParserJSON parser;
     	@Autowired
     	private MongoClient mongoClient;
     	private MongoDatabase db;
     	private String dumpType;
     	@Override
     	protected boolean canResolvePid(final String pidType) {
     		return (pidType != null) && ("doi".equals(pidType.toLowerCase().trim()) || "handle".equals(pidType.toLowerCase().trim()));
-...
                 if (record.getPid() == null) return null;
     			return record;
     		} catch (Throwable e) {
     			log.error("Error on resolve pid  " + pid, e);
     			log.error("¯\\_(ツ)_/¯ Error on resolve pid  " + pid, e);
+    		}
     		return null;
+    	}
     	private String retrieveCrossRefFromDump(final String pid) {
     		if (dumpType.equalsIgnoreCase("mongo")){
     			return retrieveCrossRefFromDumpMongo(pid);
+    		}
     		else if (dumpType.equalsIgnoreCase("ES")){
     			return retrieveCrossRefFromDumpES(pid);
+    		}
     		throw new  RuntimeException("incorrect dump Type expected [mongo, ES] found: "+dumpType );
+    	}
     		final String response = requestURL("http://ip-90-147-167-137.ct1.garrservices.it:9200/crossref/item/" + pid.replaceAll("/","%2F"));
     		return response;
     	private String retrieveCrossRefFromDumpES(final String pid) {
     		return  requestURL("http://ip-90-147-167-137.ct1.garrservices.it:9200/crossref/item/" + pid.replaceAll("/","%2F"));
+    	}
     	private String retrieveCrossRefFromDumpMongo(final String pid) {
     		if (db == null) {
     			db = mongoClient.getDatabase("crossRef");
+    		}
     		final MongoCollection<Document> crossRef = db.getCollection("dump");
     		DBObject query = QueryBuilder.start("_id").is(pid).get();
     		FindIterable<Document> documents = crossRef.find((Bson) query).limit(1);
     		MongoCursor<Document> iterator = documents.iterator();
     		if (iterator.hasNext()){
     			return iterator.next().toJson();
+    		}
     		return null;
+    	}
     	public String getDumpType() {
     		return dumpType;
+    	}
     	@Required
     	public void setDumpType(String dumpType) {
     		this.dumpType = dumpType;
+    	}
+    }

     		if (record == null) return null;
     		JsonElement jElement = new JsonParser().parse(record);
     		JsonElement source = jElement.getAsJsonObject().get("_source");
     		if (source== null || !source.isJsonObject())
     		JsonElement source = null;
     		if (jElement.getAsJsonObject().has("_source")) {
     			source = jElement.getAsJsonObject().get("_source");
     			if (source == null || !source.isJsonObject())
     				return null;
+    		}
     		else if(jElement.getAsJsonObject().has("DOI")){
     			source = jElement;
     		} else {
     			return null;
+    		}
     		final JsonObject message = source.getAsJsonObject();
             DLIResolvedObject currentObject = new DLIResolvedObject();

modules/dnet-dli/trunk/src/main/resources/eu/dnetlib/dli/resolver/applicationContext-dli-resolver.properties
1	1	services.dli.resolver.store.DatabaseName=resolverStore
2		services.dli.resolver.store.CollectionName=resolverCollection
	2	services.dli.resolver.store.CollectionName=resolverCollection
	3	services.dli.resolver.crossRef.dump=mongo

     		<property name="cache" ref="dliResolverCache"/>
     	</bean>
         <bean id="crossrefResolver" class="eu.dnetlib.dli.resolver.CrossrefResolver" p:order="0" p:availableOffline="true">
     		<property name="cache" ref="dliResolverCache"/>
         <bean id="crossrefResolver" class="eu.dnetlib.dli.resolver.CrossrefResolver" p:order="0" p:availableOffline="true"
     		p:dumpType="${services.dli.resolver.crossRef.dump}" p:cache-ref="dliResolverCache">
     	</bean>
     	<bean id="dliOfflineResolver" class="eu.dnetlib.dli.resolver.DLIOfflineResolver" p:order="0" p:availableOffline="true">

                 	<PARAM name="unknownMdstoreId"                  description="Store for unknown records"             required="true" managedBy="system" category="MDSTORE_ID"/>
                 	<PARAM name="cleanTransformationRuleId"         description="Transformation Rule Identifier"        required="true" managedBy="user"   category="TRANSFORMATION_RULE_ID" type="string" function="listProfiles('TransformationRuleDSResourceType', '//TITLE', 'DLI:')"/>
                 	<PARAM description="Type of Transformation" function="validValues(['simpleTransform', 'transformAndUnpack'])" managedBy="user" name="typeOfTransform" required="true" type="string">simpleTransform</PARAM>
                 	<PARAM description="Incremental Transformation" function="validValues(['INCREMENTAL', 'REFRESH'])" managedBy="user" name="collectionMode" required="false" type="string">REFRESH</PARAM>
                 	<PARAM description="Resolving offline" function="validValues(['false', 'true'])" managedBy="user" name="offlineResolving" required="true" type="string">false</PARAM>
                 	<PARAM name="from_date"                         description="Start Date of Harvesting"              required="false" managedBy="user"   category="COLLECTION" type="string"/>
                 </PARAMETERS>
                 <WORKFLOW>
                     <NODE isStart="true" name="collection" type="LaunchWorkflowTemplate">
-...
                         			<ENTRY key="dsId"           value="$dsId$" />
      								<ENTRY key="interface"      value="$interface$" />
     	 							<ENTRY key="collMdstoreId"  ref="collMdstoreId" />
     	 							<ENTRY key="collectionMode"  ref="collectionMode" />
     	 							<ENTRY key="from_date"      ref="from_date" />
                         		</MAP>
                         	</PARAM>
                         </PARAMETERS>
-...
      								<ENTRY key="collMdstoreId"          ref="collMdstoreId" />
      								<ENTRY key="cleanMdstoreId"         ref="cleanMdstoreId" />
      								<ENTRY key="cleanRuleId"            ref="cleanTransformationRuleId" />
      								<ENTRY key="typeOfTransform"        ref="typeOfTransform" />
      								<ENTRY key="collectionMode"         ref="collectionMode" />
      								<ENTRY key="typeOfTransform"         ref="typeOfTransform" />
      							</MAP>
      						</PARAM>
-...
                                         <ENTRY key="collMdstoreId"          ref="collMdstoreId" />
                                         <ENTRY key="cleanMdstoreId"         ref="cleanMdstoreId" />
                                         <ENTRY key="offlineResolving"       ref="offlineResolving" />
                                         <ENTRY key="collectionMode"         ref="collectionMode" />
                                     </MAP>
                                 </PARAM>
                             </PARAMETERS>

                 <IMPORTED/>
                 <SCRIPT>
                     <TITLE>DLI: ENA transform</TITLE>
                     <CODE><![CDATA[<xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform">
         <xsl:template match="@*|node()">
                     <CODE><![CDATA[<xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform"
                     xmlns:dri="http://www.driver-repository.eu/namespace/dri"
                     xmlns:dc="http://purl.org/dc/elements/1.1/"
                     xmlns:oaf="http://namespace.dnet.eu/oaf" version="2.0" exclude-result-prefixes="xsl">
         <xsl:template match="/">
             <xsl:apply-templates mode="dli"></xsl:apply-templates>
         </xsl:template>
         <xsl:template match="@*|node()" mode="dli">
             <xsl:copy>
                 <xsl:apply-templates select="@*|node()"/>
             </xsl:copy>
                 <xsl:apply-templates select="@*|node()" mode="dli"/>
             </xsl:copy>elasticsearchFeedDliSummaryDataJob
         </xsl:template>
         <xsl:template match="*[local-name()='about']" mode="dli">
             <oaf:about>
                 <oaf:datainfo>
                     <oaf:completionStatus>complete</oaf:completionStatus>
                     <xsl:choose>
                         <xsl:when test="//dc:type='publication'">
                             <oaf:collectedFrom completionStatus="incomplete" id="r3d100010527" name="European Nucleotide Archive" />
                             <oaf:resolvedFrom completionStatus="complete" id="dli_________::pubmed" name="PubMed"/>
                             <oaf:provisionMode>resolved</oaf:provisionMode>
                         </xsl:when>
                         <xsl:otherwise>
                             <oaf:collectedFrom completionStatus="complete" id="r3d100010527" name="European Nucleotide Archive" />
                             <oaf:provisionMode>collected</oaf:provisionMode>
                         </xsl:otherwise>
                     </xsl:choose>
                 </oaf:datainfo>
             </oaf:about>
         </xsl:template>
     </xsl:stylesheet>
     ]]></CODE>
                 </SCRIPT>

     				<INTERFACE active="true" compliance="dli" contentDescription="metadata" id="api_________::opendoar____::2367::0"
     				           label="Links provider" typology="links::provider" removable="false">
     					<ACCESS_PROTOCOL format="oai_datacite" set="BL.CCDC">oai</ACCESS_PROTOCOL>
     					<BASE_URL>http://oai.datacite.org/oai</BASE_URL>
     					<BASE_URL>https://oai.datacite.org/oai</BASE_URL>
     					<INTERFACE_EXTRA_FIELD name="last_collection_date"/>
     					<INTERFACE_EXTRA_FIELD name="metadata_identifier_path">//*[local-name()='header']/*[local-name()='identifier']
     					</INTERFACE_EXTRA_FIELD>

     			<INTERFACES>
     				<INTERFACE active="true" compliance="dli" contentDescription="metadata" id="api_________::datacite::0"
     				           label="Links provider" typology="links::provider" removable="false">
     					<ACCESS_PROTOCOL>mongoDump</ACCESS_PROTOCOL>
     					<BASE_URL>/home/sandro/mongo_backup/datacite3000.json</BASE_URL>
     					<ACCESS_PROTOCOL format="oai_datacite">oai</ACCESS_PROTOCOL>
     					<BASE_URL>https://oai.datacite.org/oai</BASE_URL>
     					<INTERFACE_EXTRA_FIELD name="last_collection_date"/>
     					<INTERFACE_EXTRA_FIELD name="metadata_identifier_path">//*[local-name()='header']/*[local-name()='identifier']
     					</INTERFACE_EXTRA_FIELD>

     				<PARAM name="cleanMdstoreId" description="Store for cleaned records" required="true" type="string"/>
                     <PARAM name="numberOfThreads" description="number of threads for resolve PID" required="true"   type="string" default="4"/>
     				<PARAM name="offlineResolving" description="decide if you want resolve offline using only the resolved item" required="true" type="boolean" default="false"/>
     				<PARAM name="collectionMode" description="Define the type of collection mode [INCREMENTAL, REFRESH]" required="true" default="REFRESH" type="string"/>
     			</PARAMETERS>
     			<WORKFLOW>
-...
     						<PARAM name="pluginName" value="dliResolverPlugin"/>
                             <PARAM name="numberOfThreads" ref="numberOfThreads"/>
                             <PARAM name="offline" ref="offlineResolving"/>
                             <PARAM name="collectionMode" ref="collectionMode"/>
                         </PARAMETERS>
     					<ARCS>
     						<ARC to="UPDATE_INFO"/>

     				<PARAM name="dsId" description="Datasource Id" required="true" type="string"/>
     				<PARAM name="interface" description="Datasource Interface" required="true" type="string"/>
     				<PARAM name="collMdstoreId" description="Store for collected records" required="true" type="string"/>
     				<PARAM name="from_date" description="From Date" required="false" type="string"/>
     				<PARAM name="collectionMode" description="Define the type of collection mode [INCREMENTAL, REFRESH]" required="true" default="REFRESH" type="string"/>
     			</PARAMETERS>
     			<WORKFLOW>
     				<NODE name="COLLECT_REFRESH" type="CollectRecords" isStart="true">
     				<NODE name="FindDateRangeForIncrementalHarvesting" type="FindDateRangeForIncrementalHarvesting"
     					  isStart="true">
     					<DESCRIPTION>Find Last execution</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="fromDateParam" ref="fromDateParam"/>
     						<PARAM name="collectionMode" ref="collectionMode"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="COLLECT"/>
     					</ARCS>
     				</NODE>
     				<NODE name="COLLECT" type="DateRangeCollectRecords">
     					<DESCRIPTION>Start Harvesting</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="datasourceId" ref="dsId"/>
     						<PARAM name="datasourceInterface" ref="interface"/>
     						<PARAM name="eprParam" value="collected_epr"/>
     						<PARAM name="from" ref="from_date"/>
     						<PARAM name="fromDateParam" ref="fromDateParam"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="MD_BUILDER"/>
-...
     						<PARAM name="datasourceInterface" ref="interface"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="STORE_REFRESH"/>
     						<ARC to="STORE"/>
     					</ARCS>
     				</NODE>
     				<NODE name="STORE_REFRESH" type="StoreMDStoreRecords">
     				<NODE name="STORE" type="StoreMDStoreRecords">
     					<DESCRIPTION>Store mdstore records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="mdId" ref="collMdstoreId"/>
     						<PARAM name="storingType" value="REFRESH"/>
     						<PARAM name="storingType" ref="collectionMode"/>
     						<PARAM name="eprParam" value="store_epr"/>
     					</PARAMETERS>
     					<ARCS>

     <?xml version="1.0" encoding="UTF-8"?>
     <RESOURCE_PROFILE>
     	<HEADER>
     		<RESOURCE_IDENTIFIER value="da3531c6-2bf6-48ab-848b-bd4c6379fd65_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>
     		<RESOURCE_TYPE value="WorkflowTemplateDSResourceType"/>
     		<RESOURCE_KIND value="WorkflowTemplateDSResources"/>
     		<RESOURCE_URI value=""/>
     		<DATE_OF_CREATION value="2001-12-31T12:00:00"/>
     		<DATE_OF_CREATION value="2018-02-12T12:49:19+01:00"/>
     	</HEADER>
     	<BODY>
     		<CONFIGURATION>
     			<PARAMETERS>
     				<PARAM name="dsId" description="Datasource Id" required="true" type="string"/>
     				<PARAM name="interface" description="Datasource Interface" required="true" type="string"/>
     				<PARAM name="collMdstoreId" description="Store for collected records" required="true" type="string"/>
     				<PARAM name="cleanMdstoreId" description="Store for cleaned records" required="true" type="string"/>
     				<PARAM name="cleanRuleId" description="Transformation Rule Id" required="true" type="string"/>
                     <PARAM name="typeOfTransform" description="Type of transformation" required="true" type="string"/>
     				<PARAM description="Datasource Id" name="dsId" required="true" type="string"/>
     				<PARAM description="Datasource Interface" name="interface" required="true" type="string"/>
     				<PARAM description="Store for collected records" name="collMdstoreId" required="true" type="string"/>
     				<PARAM description="Store for cleaned records" name="cleanMdstoreId" required="true" type="string"/>
     				<PARAM description="Transformation Rule Id" name="cleanRuleId" required="true" type="string"/>
     				<PARAM description="Type of transformation" name="typeOfTransform" required="true" type="string"/>
     				<PARAM name="collectionMode" description="Define the type of collection mode [INCREMENTAL, REFRESH]" required="true" default="REFRESH" type="string"/>
     			</PARAMETERS>
     			<WORKFLOW>
     				<NODE name="fetchOriginals" type="FetchMDStoreRecords" isStart="true">
     				<NODE isStart="true" name="setIncrementalTransformation" type="IncrementalTransformation">
     					<DESCRIPTION>Set incremental transformation</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="transformationType" ref="collectionMode"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="fetchOriginals"/>
     					</ARCS>
     				</NODE>
     				<NODE name="fetchOriginals" type="FetchMDStoreRecords">
     					<DESCRIPTION>Fetch records from MDStore</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="mdId" ref="collMdstoreId"/>
     						<PARAM name="eprParam" value="orig_epr"/>
     					</PARAMETERS>
     					<ARCS>
                             <ARC to="transformType"/>
     						<ARC to="transformType"/>
     					</ARCS>
     				</NODE>
                     <NODE name="transformType" type="Selection">
                         <DESCRIPTION>Clean original records</DESCRIPTION>
                         <PARAMETERS>
                             <PARAM name="selection" ref="typeOfTransform"/>
                         </PARAMETERS>
                         <ARCS>
                             <ARC name="simpleTransform" to="simpleTransform"/>
                             <ARC name="transformAndUnpack" to="transformAndUnpack"/>
                         </ARCS>
                     </NODE>
                     <NODE name="simpleTransform" type="Transform">
     				<NODE name="transformType" type="Selection">
     					<DESCRIPTION>Clean original records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="selection" ref="typeOfTransform"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC name="simpleTransform" to="simpleTransform"/>
     						<ARC name="transformAndUnpack" to="transformAndUnpack"/>
     					</ARCS>
     				</NODE>
     				<NODE name="simpleTransform" type="Transform">
     					<DESCRIPTION>Clean original records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="ruleId" ref="cleanRuleId"/>
     						<PARAM name="inputEprParam" value="orig_epr"/>
     						<PARAM name="outputEprParam" value="clean_epr"/>
     					</PARAMETERS>
     					<ARCS>
                             <ARC to="storeSimpleRecords"/>
     						<ARC to="storeSimpleRecords"/>
     					</ARCS>
     				</NODE>
                     <NODE name="storeSimpleRecords" type="StoreMDStoreRecords">
                         <DESCRIPTION>Store mdtore records</DESCRIPTION>
                         <PARAMETERS>
                             <PARAM name="mdId" ref="cleanMdstoreId"/>
                             <PARAM name="storingType" value="REFRESH"/>
                             <PARAM name="eprParam" value="clean_epr"/>
                         </PARAMETERS>
                         <ARCS>
                             <ARC to="UPDATE_INFO"/>
                         </ARCS>
                     </NODE>
                     <NODE name="transformAndUnpack" type="Transform">
                         <DESCRIPTION>Clean original records</DESCRIPTION>
                         <PARAMETERS>
                             <PARAM name="ruleId" ref="cleanRuleId"/>
                             <PARAM name="inputEprParam" value="orig_epr"/>
                             <PARAM name="outputEprParam" value="pack_epr"/>
                         </PARAMETERS>
                         <ARCS>
                             <ARC to="unpackRecord"/>
                         </ARCS>
                     </NODE>
                     <NODE name="unpackRecord" type="Unpack">
                         <DESCRIPTION>Unpack transformed records</DESCRIPTION>
                         <PARAMETERS>
                             <PARAM name="xpath" value="//*[local-name()='record']"/>
                             <PARAM name="inputEprParam" value="pack_epr"/>
                             <PARAM name="outputEprParam" value="clean_epr"/>
                         </PARAMETERS>
                         <ARCS>
                             <ARC to="storeUnpackedRecords"/>
                         </ARCS>
                     </NODE>
                     <NODE name="storeUnpackedRecords" type="StoreMDStoreRecords">
     				<NODE name="storeSimpleRecords" type="StoreMDStoreRecords">
     					<DESCRIPTION>Store mdtore records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="mdId" ref="cleanMdstoreId"/>
-...
     						<ARC to="UPDATE_INFO"/>
     					</ARCS>
     				</NODE>
     				<NODE name="transformAndUnpack" type="Transform">
     					<DESCRIPTION>Clean original records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="ruleId" ref="cleanRuleId"/>
     						<PARAM name="inputEprParam" value="orig_epr"/>
     						<PARAM name="outputEprParam" value="pack_epr"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="unpackRecord"/>
     					</ARCS>
     				</NODE>
     				<NODE name="unpackRecord" type="Unpack">
     					<DESCRIPTION>Unpack transformed records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="xpath" value="//*[local-name()='record']"/>
     						<PARAM name="inputEprParam" value="pack_epr"/>
     						<PARAM name="outputEprParam" value="clean_epr"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="storeUnpackedRecords"/>
     					</ARCS>
     				</NODE>
     				<NODE name="storeUnpackedRecords" type="StoreMDStoreRecords">
     					<DESCRIPTION>Store mdtore records</DESCRIPTION>
     					<PARAMETERS>
     						<PARAM name="mdId" ref="cleanMdstoreId"/>
     						<PARAM name="storingType" value="REFRESH"/>
     						<PARAM name="eprParam" value="clean_epr"/>
     					</PARAMETERS>
     					<ARCS>
     						<ARC to="UPDATE_INFO"/>
     					</ARCS>
     				</NODE>
     				<NODE name="UPDATE_INFO" type="MDStoreToApiExtraField">
     					<DESCRIPTION>Update datasouce API extra fields</DESCRIPTION>
     					<PARAMETERS>
-...
     			</WORKFLOW>
     		</CONFIGURATION>
     	</BODY>
     </RESOURCE_PROFILE>
     </RESOURCE_PROFILE>

     <?xml version="1.0" encoding="UTF-8"?>
     <oai:record xmlns:oai="http://www.openarchives.org/OAI/2.0/"
                 xmlns="http://namespace.openaire.eu/">
         <oai:header xmlns="">
             <dri:objIdentifier xmlns:dri="http://www.driver-repository.eu/namespace/dri">
                 r3d100010464::a57f57564a6f58554f6dad6a954bbf55
             </dri:objIdentifier>
             <dri:recordIdentifier xmlns:dri="http://www.driver-repository.eu/namespace/dri">1959.1/468446::hdl
             </dri:recordIdentifier>
             <dri:dateOfCollection xmlns:dri="http://www.driver-repository.eu/namespace/dri">2017-09-22T08:57:59.857+02:00
             </dri:dateOfCollection>
             <dri:repositoryId xmlns:dri="http://www.driver-repository.eu/namespace/dri">
                 ands_UmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZXMvUmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZVR5cGU=
             </dri:repositoryId>
             <dri:datasourceprefix xmlns:dri="http://www.driver-repository.eu/namespace/dri">r3d100010464
             </dri:datasourceprefix>
     <oai:record xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                 xmlns:oai="http://www.openarchives.org/OAI/2.0/"
                 xmlns:dri="http://www.driver-repository.eu/namespace/dri">
         <oai:header>
             <dri:objIdentifier xmlns:oaf="http://namespace.dnet.eu/oaf"
                                xmlns:datacite="http://datacite.org/schema/kernel-3">r3d100010527::00000648e410cb638823564cf5fdc885</dri:objIdentifier>
             <dri:recordIdentifier xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">http://www.ebi.ac.uk/ena/data/search/?query=11118631&amp;result=sequence_release::url</dri:recordIdentifier>
             <dri:dateOfCollection xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">2017-09-18T14:07:46.84+02:00</dri:dateOfCollection>
             <dri:repositoryId xmlns:oaf="http://namespace.dnet.eu/oaf"
                               xmlns:datacite="http://datacite.org/schema/kernel-3">dli_________::r3d100010527</dri:repositoryId>
             <dri:datasourceprefix xmlns:oaf="http://namespace.dnet.eu/oaf"
                                   xmlns:datacite="http://datacite.org/schema/kernel-3">r3d100010527</dri:datasourceprefix>
         </oai:header>
         <metadata xmlns="">
         <metadata xmlns:dc="http://purl.org/dc/elements/1.1/"
                   xmlns:oaf="http://namespace.dnet.eu/oaf"
                   xmlns:datacite="http://datacite.org/schema/kernel-3">
             <resource xmlns="http://datacite.org/schema/kernel-3"
                       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                       xsi:schemaLocation="http://datacite.org/schema/kernel-3 http://schema.datacite.org/meta/kernel-3/metadata.xsd">
                 <identifier xmlns="" identifierType="hdl">1959.1/468446</identifier>
                 <titles xmlns="">
                     <title>Clinician practices while using health information system security in Australian public
                         hospitals: data
                     </title>
                 <identifier identifierType="url">http://www.ebi.ac.uk/ena/data/search/?query=11118631&amp;result=sequence_release</identifier>
                 <titles>
                     <title>Nucleotide sequences (Release) (1/2) of "Molecular analysis of a t(7;14)(q35;q32) chromosome translocation in a T cell leukemia of a patient with ataxia telangiectasia." </title>
                 </titles>
                 <publisher xmlns="">Monash University</publisher>
                 <dates xmlns="">
                     <date dateType="Collected">2012-11-28</date>
                 <dates>
                     <date dateType="Collected">2000-04-21</date>
                 </dates>
                 <creators xmlns="">
                     <creator>
                         <creatorName>Dr Juanita Fernando</creatorName>
                     </creator>
                 </creators>
                 <resourceType xmlns="" resourceTypeGeneral="Dataset">Dataset</resourceType>
                 <!--<relatedIdentifiers xmlns="">-->
                     <!--<relatedIdentifier entityType="                     publication                 "-->
                                        <!--inverseRelationType="hasAssociationWith"-->
                                        <!--relatedIdentifierType="dnet"-->
                                        <!--relationType="hasAssociationWith">r3d100010255::d5ad02b122901f0d9d59f8348768d0f9-->
                     <!--</relatedIdentifier>-->
                 <!--</relatedIdentifiers>-->
                 <descriptions>
                     <description>Catharanthus roseus 2C-methyl-D-erythritol 2,4-cyclodiphosphate synthase (MECS) mRNA, complete cds.</description>
                 </descriptions>
                 <subjects>
                     <subject subjectScheme="organism">Catharanthus roseus</subject>
                     <subject subjectScheme="codon_start">1</subject>
                     <subject subjectScheme="gene">MECS</subject>
                     <subject subjectScheme="product">2C-methyl-D-erythritol 2,4-cyclodiphosphate synthase</subject>
                     <subject subjectScheme="note">YgbB protein</subject>
                     <subject subjectScheme="protein_id">AAF65155.1</subject>
                     <subject subjectScheme="translation">MAMATSFYCSTAIPSKKTNQNRENFLCSPVGGSKTTPSYIRLSTRQSRTLSLVVSAAASGAAVEAEPKFAAVTPSKILSF
                         RVGHGFDLHRLEPGYPLIIGGINIPHDRGCEAHSDGDVLLHCVVDAILGALGLPDIGQIFPDTDPKWKGAPSSVFIKEAV
                         RLMDEAGYELGNLDATLILQRPKVSPHKEAIRQNLCQLLGADPCVVNLKAKTHEKVDSLGENRSIAAHTVVLLMRK</subject>
                 </subjects>
                 <resourceType xmlns="" resourceTypeGeneral="dataset">dataset</resourceType>
                 <oaf:relatedIdentifier entityType="publication" inverseRelationType="isRelatedTo"
                                        relatedIdentifierType="dnet"
                                        relationType="isRelatedTo">r3d100010527::f89cc26f592efddee8437a6a73a6ab5a</oaf:relatedIdentifier>
             </resource>
         </metadata>
         <oaf:about xmlns:oaf="http://namespace.dnet.eu/oaf" xmlns="">
         <oaf:about xmlns:oaf="http://namespace.dnet.eu/oaf">
             <oaf:datainfo>
                 <oaf:collectedFrom completionStatus="complete" id="dli_________::r3d100010464"
                                    name="Australian National Data Service"/>
                 <oaf:completionStatus>complete</oaf:completionStatus>
                 <oaf:provisionMode>collected</oaf:provisionMode>
                 <oaf:collectedFrom completionStatus="complete" id="dli_________::datacite"
                                    name="Datasets in Datacite"/>
             </oaf:datainfo>
         </oaf:about>
     </oai:record>

     <?xml version="1.0" encoding="UTF-8"?>
     <record xmlns:oaf="http://namespace.dnet.eu/oaf"
             xmlns:dri="http://www.driver-repository.eu/namespace/dri"
             xmlns:dc="http://purl.org/dc/elements/1.1/">
         <oai:header xmlns:oai="http://www.openarchives.org/OAI/2.0/"
                     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
             <dri:objIdentifier>pubmed______::000087d2ea077b8526bbc3c1436737ac</dri:objIdentifier>
             <dri:resolvedDate>2017-10-31T16:52:57.209</dri:resolvedDate>
             <dri:recordIdentifier>EuropePMC:26375944</dri:recordIdentifier>
             <dri:dateOfCollection>2017-10-31T16:51:58.293+01:00</dri:dateOfCollection>
             <dri:repositoryId>pbm_UmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZXMvUmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZVR5cGU=</dri:repositoryId>
             <dri:datasourceprefix>pubmed______</dri:datasourceprefix>
     <oai:record xmlns:oaf="http://namespace.dnet.eu/oaf"
                 xmlns:datacite="http://datacite.org/schema/kernel-3"
                 xmlns:oai="http://www.openarchives.org/OAI/2.0/"
                 xmlns:dri="http://www.driver-repository.eu/namespace/dri">
         <oai:header>
             <dri:objIdentifier>dli_resolver::b65cd86a86f67af94a3a25452b8ab30d</dri:objIdentifier>
             <dri:recordIdentifier>b65cd86a86f67af94a3a25452b8ab30d</dri:recordIdentifier>
             <dri:dateOfCollection>2017-12-20T15:37:59.445+01:00</dri:dateOfCollection>
             <dri:repositoryId>dli_resolver</dri:repositoryId>
             <dri:datasourceprefix>dli_resolver</dri:datasourceprefix>
         </oai:header>
         <metadata>
             <oaf:pid type="pmid">26375944</oaf:pid>
             <dc:identifier>http://www.ncbi.nlm.nih.gov/pubmed/26375944</dc:identifier>
             <dc:title>Reclassification of Saccharomycodes sinensis, Proposal of Yueomyces sinensis gen. nov., comb. nov. within Saccharomycetaceae (Saccharomycetales, Saccharomycotina).</dc:title>
             <dc:creator>Long Wang</dc:creator>
             <dc:creator>Marizeth Groenewald</dc:creator>
             <dc:creator>Qi-Ming Wang</dc:creator>
             <dc:creator>Teun Boekhout</dc:creator>
             <dc:date>2015-09-16</dc:date>
         <metadata xmlns:dc="http://purl.org/dc/elements/1.1/">
             <oaf:pid type="doi">10.1002/jame.20038</oaf:pid>
             <dc:identifier>http://dx.doi.org/10.1002/jame.20038</dc:identifier>
             <dc:title>"Climate and carbon cycle changes from 1850 to 2100 in MPI-ESM simulations for the Coupled Model Intercomparison Project phase 5"</dc:title>
             <dc:creator>Giorgetta Marco A.</dc:creator>
             <dc:creator>Jungclaus Johann</dc:creator>
             <dc:creator>Reick Christian H.</dc:creator>
             <dc:creator>Legutke Stephanie</dc:creator>
             <dc:creator>Bader Jürgen</dc:creator>
             <dc:creator>Böttinger Michael</dc:creator>
             <dc:creator>Brovkin Victor</dc:creator>
             <dc:creator>Crueger Traute</dc:creator>
             <dc:creator>Esch Monika</dc:creator>
             <dc:creator>Fieg Kerstin</dc:creator>
             <dc:creator>Glushak Ksenia</dc:creator>
             <dc:creator>Gayler Veronika</dc:creator>
             <dc:creator>Haak Helmuth</dc:creator>
             <dc:creator>Hollweg Heinz-Dieter</dc:creator>
             <dc:creator>Ilyina Tatiana</dc:creator>
             <dc:creator>Kinne Stefan</dc:creator>
             <dc:creator>Kornblueh Luis</dc:creator>
             <dc:creator>Matei Daniela</dc:creator>
             <dc:creator>Mauritsen Thorsten</dc:creator>
             <dc:creator>Mikolajewicz Uwe</dc:creator>
             <dc:creator>Mueller Wolfgang</dc:creator>
             <dc:creator>Notz Dirk</dc:creator>
             <dc:creator>Pithan Felix</dc:creator>
             <dc:creator>Raddatz Thomas</dc:creator>
             <dc:creator>Rast Sebastian</dc:creator>
             <dc:creator>Redler Rene</dc:creator>
             <dc:creator>Roeckner Erich</dc:creator>
             <dc:creator>Schmidt Hauke</dc:creator>
             <dc:creator>Schnur Reiner</dc:creator>
             <dc:creator>Segschneider Joachim</dc:creator>
             <dc:creator>Six Katharina D.</dc:creator>
             <dc:creator>Stockhause Martina</dc:creator>
             <dc:creator>Timmreck Claudia</dc:creator>
             <dc:creator>Wegner Jörg</dc:creator>
             <dc:creator>Widmann Heinrich</dc:creator>
             <dc:creator>Wieners Karl-H.</dc:creator>
             <dc:creator>Claussen Martin</dc:creator>
             <dc:creator>Marotzke Jochem</dc:creator>
             <dc:creator>Stevens Bjorn</dc:creator>
             <dc:date/>
             <dc:description/>
             <dc:type>publication</dc:type>
             <oaf:relatedIdentifier entityType="dataset" inverseRelationType="isRelatedTo"
                                    relatedIdentifierType="dnet"
                                    relationType="isRelatedTo">dli_resolver::d90bf4e9a54d2dffc53b5ec6ce7c2dd6</oaf:relatedIdentifier>
             <oaf:relatedIdentifier entityType="dataset" inverseRelationType="isRelatedTo"
                                    relatedIdentifierType="dnet"
                                    relationType="isRelatedTo">dli_resolver::17a9492cc1632b6b95c6e7067d145162</oaf:relatedIdentifier>
             <oaf:relatedIdentifier entityType="dataset" inverseRelationType="isRelatedTo"
                                    relatedIdentifierType="dnet"
                                    relationType="isRelatedTo">dli_resolver::12e836f393a607172c54013f7a4d9816</oaf:relatedIdentifier>
             <oaf:relatedIdentifier entityType="dataset" inverseRelationType="isRelatedTo"
                                    relatedIdentifierType="dnet"
                                    relationType="isRelatedTo">dli_resolver::c7edca93b5d2524986f2dc1b2c55f600</oaf:relatedIdentifier>
             <dc:publisher>Wiley-Blackwell</dc:publisher>
         </metadata>
         <oaf:about>
         <oaf:about xmlns:dc="http://purl.org/dc/elements/1.1/">
             <oaf:datainfo>
                 <oaf:completionStatus>complete</oaf:completionStatus>
                 <oaf:collectedFrom completionStatus="complete" id="pubmed______" name="PubMed"/>
                 <oaf:resolvedFrom id="dli_________::crossref" name="Crossref" completionStatus="complete"/>
             </oaf:datainfo>
         </oaf:about>
     </record>
     </oai:record>

     import java.io.IOException;
     import org.apache.commons.logging.Log;
     import org.apache.commons.logging.LogFactory;
     import org.apache.hadoop.io.Text;
     import org.apache.hadoop.mapreduce.Mapper;
-...
      */
     public class MindistSearchMapper extends Mapper<Text, VertexWritable, Text, VertexWritable> {
     	private static final Log log = LogFactory.getLog(MindistSearchMapper.class);
     	private boolean debug = false;
     	@Override
     	protected void setup(Mapper.Context context) throws IOException, InterruptedException {
     		super.setup(context);
     		debug = context.getConfiguration().getBoolean("mindist_DEBUG", false);
     		log.info("debug mode: " + debug);
+    	}
     	@Override
     	protected void map(Text key, VertexWritable value, Context context) throws IOException, InterruptedException {
     		context.write(key, value);
     		emit(key, value, context);
     		if (value.isActivated()) {
     			VertexWritable writable = new VertexWritable();
     			for (Text neighborVertex : value.getEdges()) {
     				if (!neighborVertex.toString().equals(value.getVertexId().toString())) {
     					writable.setVertexId(value.getVertexId());
     					writable.setEdges(null);
     					context.write(neighborVertex, writable);
     			VertexWritable vertex = new VertexWritable();
     			for (Text edge : value.getEdges()) {
     				if (!edge.toString().equals(value.getVertexId().toString())) {
     					vertex.setVertexId(value.getVertexId());
     					vertex.setEdges(null);
     					emit(edge, vertex, context);
+    				}
+    			}
+    		}
+    	}
     	private void emit(final Text key, final VertexWritable vertex, final Context context) throws IOException, InterruptedException {
     		context.write(key, vertex);
     		if (debug) {
     			log.info(vertex.toJSON());
+    		}
+    	}
+    }

     	private static final Log log = LogFactory.getLog(MindistSearchReducer.class);
     	public static final String UPDATE_COUNTER = "UpdateCounter";
     	public static final String SKIPPED = "SKIPPED";
     	public static final String UPDATED = "UPDATED";
     	private boolean depthOne;
     	private boolean debug = false;
-...
     			depthOne = true;
+    		}
     		try {
     			debug = Boolean.valueOf(context.getConfiguration().get("mindist_DEBUG"));
     		} catch(Throwable e) {
     			debug = false;
+    		}
     		debug = context.getConfiguration().getBoolean("mindist_DEBUG", false);
     		log.info("debug mode: " + debug);
+    	}
-...
     		VertexWritable realVertex = null;
     		Text currentMinimalKey = null;
     		boolean foundEdges = false;
     		//boolean foundEdges = false;
     		if (depthOne) {
     			for (VertexWritable vertex : values) {
     				if (!vertex.isMessage()) {
     					//log.info(String.format("found vertex with edges: %s", key.toString()));
     					realVertex = vertex.clone();
     					foundEdges = true;
+    				}
+    			}
     			if (realVertex == null) {
     				throw new IllegalStateException(String.format("foundEdges: %s, invalid input, key: '%s'", foundEdges, key.toString()));
     				context.getCounter(UPDATE_COUNTER, SKIPPED).increment(1);
     				return;
+    			}
     			realVertex.setActivated(true);
-...
     				realVertex.setVertexId(key);
+    			}
     			context.getCounter("UpdateCounter", "UPDATED").increment(1);
     			context.getCounter(UPDATE_COUNTER, UPDATED).increment(1);
     		} else {
     			for (VertexWritable vertex : values) {
     				if (!vertex.isMessage()) {
-...
+    				}
+    			}
     			if (realVertex == null) {
     				context.getCounter(UPDATE_COUNTER, SKIPPED).increment(1);
     				return;
+    			}
     			if (currentMinimalKey != null && currentMinimalKey.compareTo(realVertex.getVertexId()) < 0) {
     				realVertex.setVertexId(currentMinimalKey);
     				realVertex.setActivated(true);
     				context.getCounter("UpdateCounter", "UPDATED").increment(1);
     				context.getCounter(UPDATE_COUNTER, UPDATED).increment(1);
     			} else {
     				realVertex.setActivated(false);
+    			}

     import java.io.IOException;
     import java.util.List;
     import java.util.stream.Collectors;
     import java.util.stream.StreamSupport;
     import com.google.common.collect.Iterables;
     import com.google.common.collect.Lists;
-...
         private DedupConfig dedupConf;
         private Ontologies ontologies;
     	@Override
     	protected void setup(final Context context) throws IOException, InterruptedException {
     		super.setup(context);
     		dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
     		System.out.println("dedup buildRoots reducer\n\nwf conf: " + dedupConf.toString());
         @Override
         protected void setup(final Context context) throws IOException, InterruptedException {
             super.setup(context);
             dedupConf = DedupConfig.load(context.getConfiguration().get(JobParams.DEDUP_CONF));
             System.out.println("dedup buildRoots reducer\n\nwf conf: " + dedupConf.toString());
     		ontologies = OntologyLoader.loadOntologies(context.getConfiguration().get(JobParams.ONTOLOGIES));
     		System.out.println("ontologies: " + ontologies.toJson(true));
             ontologies = OntologyLoader.loadOntologies(context.getConfiguration().get(JobParams.ONTOLOGIES));
             System.out.println("ontologies: " + ontologies.toJson(true));
+    	}
+        }
     	@Override
     	protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {
         @Override
         protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {
             // ensures we're dealing with a root, otherwise returns
             if (!isRoot(key.toString())) {
                 System.err.println("aborting DedupBuildRootsReducer, found non-root key: " + key);
                 context.getCounter("DedupBuildRootsReducer", "aborted").increment(1);
                 return;
+            }
     		// ensures we're dealing with a root, otherwise returns
     		if (!isRoot(key.toString())) {
     			System.err.println("aborting DedupBuildRootsReducer, found non-root key: " + key);
     			context.getCounter("DedupBuildRootsReducer", "aborted").increment(1);
     			return;
+    		}
     		final byte[] rowkey = Bytes.toBytes(key.toString());
     		final List<DNGF> entities = Lists.newArrayList();
             final byte[] rowkey = Bytes.toBytes(key.toString());
             final List<DNGF> entities = Lists.newArrayList();
     		for (final DNGF oaf : toDNGF(values)) {
     			switch (oaf.getKind()) {
     			case entity:
     				entities.add(oaf);
     				break;
     			case relation:
     				handleRels(context, rowkey, oaf, true);
     				break;
     			default:
     				break;
+    			}
+    		}
     		// build and emit the root body
     		final DNGF.Builder builder = DNGFEntityMerger.merge(dedupConf, key.toString(), entities);
     		if (entities.size() < JobParams.MAX_COUNTERS) {
     			context.getCounter(dedupConf.getWf().getEntityType() + " root group size", lpad(entities.size())).increment(1);
     		} else {
     			context.getCounter(dedupConf.getWf().getEntityType() + " root group size", "> " + JobParams.MAX_COUNTERS).increment(1);
+    		}
             for (final DNGF oaf : toDNGF(values)) {
                 switch (oaf.getKind()) {
                     case entity:
                         entities.add(oaf);
                         break;
                     case relation:
                         handleRels(context, rowkey, oaf, true);
                         break;
                     default:
                         break;
+                }
+            }
             // build and emit the root body
             final DNGF.Builder builder = DNGFEntityMerger.merge(dedupConf, key.toString(), entities);
             if (entities.size() < JobParams.MAX_COUNTERS) {
                 context.getCounter(dedupConf.getWf().getEntityType() + " root group size", lpad(entities.size())).increment(1);
             } else {
                 context.getCounter(dedupConf.getWf().getEntityType() + " root group size", "> " + JobParams.MAX_COUNTERS).increment(1);
+            }
     		final DNGF dngf = builder.build();
     		final DNGFEntity entity = dngf.getEntity();
     		try {
             final DNGF dngf = builder.build();
             final DNGFEntity entity = dngf.getEntity();
     			context.write(new ImmutableBytesWritable(Bytes.toBytes(entity.getId())), asPut(dngf));
     		} catch (Throwable e) {
     			System.out.println("Exception dngf = " + dngf.toString());
     			context.getCounter(entity.getType().toString(), e.getClass().getName()).increment(1);
+    		}
             try {
     		context.getCounter(entity.getType().toString(), "root").increment(1);
                 context.write(new ImmutableBytesWritable(Bytes.toBytes(entity.getId())), asPut(dngf));
             } catch (Throwable e) {
                 System.out.println("Exception dngf = " + dngf.toString());
                 context.getCounter(entity.getType().toString(), e.getClass().getName()).increment(1);
+            }
     		// add person rels TODO: remove this hack
     		// context.getCounter("hack", "personResult out").increment(personMap.size());
             context.getCounter(entity.getType().toString(), "root").increment(1);
+    	}
             // add person rels TODO: remove this hack
             // context.getCounter("hack", "personResult out").increment(personMap.size());
     	private Iterable<DNGF> toDNGF(final Iterable<ImmutableBytesWritable> values) {
     		return Iterables.transform(values, ibw -> DNGFDecoder.decode(ibw.copyBytes()).getDNGF());
+    	}
+        }
     	private void handleRels(final Context context, final byte[] rowkey, final DNGF rel, final boolean hack) throws IOException, InterruptedException {
         private Iterable<DNGF> toDNGF(final Iterable<ImmutableBytesWritable> values) {
             return StreamSupport.stream(values.spliterator(), false).map(ibw -> DNGFDecoder.decode(ibw.copyBytes()).getDNGF()).collect(Collectors.toList());
+        }
     		if (hack && checkHack(new String(rowkey), rel)) {
     			context.getCounter("hack", "personResult in").increment(1);
     		} else {
         private void handleRels(final Context context, final byte[] rowkey, final DNGF rel, final boolean hack) throws IOException, InterruptedException {
             // emit relation from the root to the related entities
             DNGFDecoder decoder = rootToEntity(rowkey, rel, context);
     			// emit relation from the root to the related entities
     			DNGFDecoder decoder = rootToEntity(rowkey, rel, context);
     			context.write(new ImmutableBytesWritable(rowkey), asPutByCollectedFrom(decoder.getDNGF()));
     			// emit relation from the related entities to the root
     			decoder = entityToRoot(rowkey, rel, context);
     			final byte[] revKey = Bytes.toBytes(decoder.relSourceId());
     			context.write(new ImmutableBytesWritable(revKey), asPutByCollectedFrom(decoder.getDNGF()));
                 //context.getCounter(FIXED_RELATION, decoder.getRelDescriptor().shortQualifier() + " [entity <-> root]").increment(2);
             if ("50|dedup_wf_001::cddddb031b9f6c85046067c0cc9ad147".equals(decoder.getDNGFRel().getSource()))
+            {
                 System.out.println(String.format("Writing relation %s    %s     %s ", decoder.getDNGFRel().getSource(),decoder.getDNGFRel().getRelType().getClassname(),decoder.getDNGFRel().getTarget()));
+            }
             context.write(new ImmutableBytesWritable(rowkey), asPutByCollectedFrom(decoder.getDNGF()));
             // emit relation from the related entities to the root
             decoder = entityToRoot(rowkey, rel, context);
             if ("50|dedup_wf_001::cddddb031b9f6c85046067c0cc9ad147".equals(decoder.getDNGFRel().getTarget()))
+            {
                 System.out.println(String.format("Writing relation %s    %s     %s ", decoder.getDNGFRel().getSource(),decoder.getDNGFRel().getRelType().getClassname(),decoder.getDNGFRel().getTarget()));
+            }
             byte[] revKey = Bytes.toBytes(decoder.relSourceId());
             context.write(new ImmutableBytesWritable(revKey), asPutByCollectedFrom(decoder.getDNGF()));
             //context.getCounter(FIXED_RELATION, decoder.getRelDescriptor().shortQualifier() + " [entity <-> root]").increment(2);
             // mark relation from the related entities to the duplicate as deleted
     		DNGFDecoder decoder = markDeleted(rel, true, context);
     		byte[] revKey = Bytes.toBytes(decoder.relSourceId());
     		context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));
             decoder = markDeleted(rel, true, context);
             revKey = Bytes.toBytes(decoder.relSourceId());
             context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));
     		// mark relation from the related entities to the duplicate as deleted
     		decoder = markDeleted(rel, false, context);
     		revKey = Bytes.toBytes(decoder.relSourceId());
     		context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));
             // mark relation from the related entities to the duplicate as deleted
             decoder = markDeleted(rel, false, context);
             revKey = Bytes.toBytes(decoder.relSourceId());
             context.write(new ImmutableBytesWritable(revKey), asPut(decoder.getDNGF()));
             context.getCounter(FIXED_RELATION, decoder.getRelDescriptor().shortQualifier() + " mark deleted [dup <-> entity]").increment(2);
+        }
     	public boolean checkHack(final String root, final DNGF oaf) {
         private boolean md5matches(final String id1, final String id2) {
             return id1.replaceAll("^.*\\:\\:", "").equals(id2.replaceAll("^.*\\:\\:", ""));
+        }
     		boolean res;
     		final String type = dedupConf.getWf().getEntityType();
         private DNGFDecoder rootToEntity(final byte[] rootRowkey, final DNGF rel, final Context context) {
             return patchRelations(rootRowkey, rel, DNGFPatch.rootToEntity, context);
+        }
     		if ((type.equals(Type.publication.toString()) || type.equals(Type.dataset.toString())) &&
     				oaf.getRel().getTargetType().equals(Type.person) && !md5matches(root, oaf.getRel().getSource())) {
         private DNGFDecoder entityToRoot(final byte[] rootRowkey, final DNGF rel, final Context context) {
             return patchRelations(rootRowkey, rel, DNGFPatch.entityToRoot, context);
+        }
     			res = true;
     		} else {
     			res = false;
+    		}
         private DNGFDecoder markDeleted(final DNGF rel, final boolean reverse, final Context context) {
             return deleteRelations(rel, reverse, context);
+        }
     		// if (root.equals("50|dedup_wf_001::92f6197ea6f16ae554755aced832fb6f")) {
     		// System.out.println("##################");
     		// System.out.println("root  : " + root);
     		// System.out.println("source: " + oaf.getRel().getSource());
     		// System.out.println("ckeck:  " + res);
     		// }
         // patches relation objects setting the source field with the root id
         private DNGFDecoder patchRelations(final byte[] rootRowkey, final DNGF rel, final DNGFPatch patchKind, final Context context) {
             final String id = new String(rootRowkey);
     		return res;
+    	}
             if ("50|dedup_wf_001::cddddb031b9f6c85046067c0cc9ad147".equals(id)) {
                 System.out.println("#########################\n\n");
+            }
     	private boolean md5matches(final String id1, final String id2) {
     		return id1.replaceAll("^.*\\:\\:", "").equals(id2.replaceAll("^.*\\:\\:", ""));
+    	}
     	private DNGFDecoder rootToEntity(final byte[] rootRowkey, final DNGF rel, final Context context) {
     		return patchRelations(rootRowkey, rel, DNGFPatch.rootToEntity, context);
+    	}
             final DNGFRelDecoder decoder = DNGFRelDecoder.decode(rel.getRel());
             final DNGF.Builder builder = DNGF.newBuilder(rel);
             builder.getDataInfoBuilder().setInferred(true).setDeletedbyinference(false);
             switch (patchKind) {
                 case rootToEntity:
                     // builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:rootToEntity)");
                     builder.getRelBuilder().setSource(new String(rootRowkey));
                     break;
     	private DNGFDecoder entityToRoot(final byte[] rootRowkey, final DNGF rel, final Context context) {
     		return patchRelations(rootRowkey, rel, DNGFPatch.entityToRoot, context);
+    	}
                 case entityToRoot:
                     builder.setRel(decoder.setClassId(getInverse(decoder, context)));
                     // builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:entityToRoot)");
                     builder.getRelBuilder().setSource(builder.getRel().getTarget());
                     builder.getRelBuilder().setTarget(new String(rootRowkey));
                     final Type sourceType = builder.getRelBuilder().getSourceType();
                     builder.getRelBuilder().setSourceType(builder.getRel().getTargetType());
                     builder.getRelBuilder().setTargetType(sourceType);
                     break;
     	private DNGFDecoder markDeleted(final DNGF rel, final boolean reverse, final Context context) {
     		return deleteRelations(rel, reverse, context);
+    	}
                 default:
                     break;
+            }
     	// patches relation objects setting the source field with the root id
     	private DNGFDecoder patchRelations(final byte[] rootRowkey, final DNGF rel, final DNGFPatch patchKind, final Context context) {
     		final DNGFRelDecoder decoder = DNGFRelDecoder.decode(rel.getRel());
     		final DNGF.Builder builder = DNGF.newBuilder(rel);
     		builder.getDataInfoBuilder().setInferred(true).setDeletedbyinference(false);
     		switch (patchKind) {
     		case rootToEntity:
     			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:rootToEntity)");
     			builder.getRelBuilder().setSource(new String(rootRowkey));
     			break;
             return DNGFDecoder.decode(builder.build());
+        }
     		case entityToRoot:
     			builder.setRel(decoder.setClassId(getInverse(decoder, context)));
     			// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots p:entityToRoot)");
     			builder.getRelBuilder().setSource(builder.getRel().getTarget());
     			builder.getRelBuilder().setTarget(new String(rootRowkey));
                 final Type sourceType = builder.getRelBuilder().getSourceType();
                 builder.getRelBuilder().setSourceType(builder.getRel().getTargetType());
                 builder.getRelBuilder().setTargetType(sourceType);
                 break;
     		default:
     			break;
+    		}
     		return DNGFDecoder.decode(builder.build());
+    	}
     	private String getInverse(final DNGFRelDecoder decoder, final Context context) {
     		final String inverse = ontologies.inverseOf(decoder.getRelDescriptor());
     		if (StringUtils.isBlank(inverse)) {
         private String getInverse(final DNGFRelDecoder decoder, final Context context) {
             final String inverse = ontologies.inverseOf(decoder.getRelDescriptor());
             if (StringUtils.isBlank(inverse)) {
                 //context.getCounter("unmapped relationship", decoder.getRelDescriptor().shortQualifier()).increment(1);
                 return "unknown";
+    		}
     		return inverse;
+    	}
+            }
             return inverse;
+        }
     	private DNGFDecoder deleteRelations(final DNGF rel, final boolean reverse, final Context context) {
     		final DNGF.Builder builder = DNGF.newBuilder(rel);
     		// builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots d: " + reverse + ")");
     		builder.getDataInfoBuilder().setDeletedbyinference(true);
         private DNGFDecoder deleteRelations(final DNGF rel, final boolean reverse, final Context context) {
             final DNGF.Builder builder = DNGF.newBuilder(rel);
             // builder.getDataInfoBuilder().setInferenceprovenance("dedup (BuildRoots d: " + reverse + ")");
             builder.getDataInfoBuilder().setDeletedbyinference(true);
     		if (reverse) {
     			final DNGFRelDecoder decoder = DNGFRelDecoder.decode(rel.getRel());
             if (reverse) {
                 final DNGFRelDecoder decoder = DNGFRelDecoder.decode(rel.getRel());
     			builder.setRel(decoder.setClassId(getInverse(decoder, context)));
     			// swap source and target
     			final String tmp = builder.getRel().getSource();
                 builder.setRel(decoder.setClassId(getInverse(decoder, context)));
                 // swap source and target
                 final String tmp = builder.getRel().getSource();
                 final Type sType = builder.getRel().getSourceType();
                 builder.getRelBuilder().setSource(builder.getRel().getTarget());
     			builder.getRelBuilder().setTarget(tmp);
                 builder.getRelBuilder().setTarget(tmp);
                 builder.getRelBuilder().setSourceType(builder.getRel().getTargetType());
                 builder.getRelBuilder().setTargetType(sType);
+            }
     		return DNGFDecoder.decode(builder.build());

Project

General

Profile

D-Net

Revision 50818

Added by Sandro La Bruzzo over 6 years ago