D-Net

services.hadoop.clients={"DM":{"oozie":"true","mapred":"true","hbase":"true"},"IIS":{"oozie":"true","mapred":"false","hbase":"false"},"GARR":{"oozie":"true","mapred":"false","hbase":"false"}}

	<bean id="GARR" class="eu.dnetlib.data.hadoop.config.ConfigurationFactory"

		  p:defaults="${services.data.hadoop.garr.properties}" />

dnet.clustername				=	GARR

##CORE-SITE

#fs.defaultFS					=	hdfs://dm-cluster-nn

#

#hadoop.security.authentication	=	simple

#hadoop.security.auth_to_local	=	DEFAULT

#

#hadoop.rpc.socket.factory.class.default	=	org.apache.hadoop.net.StandardSocketFactory

#

##HBASE-SITE

#hbase.rootdir					=	hdfs://dm-cluster-nn/hbase

#

#hbase.security.authentication	=	simple

#zookeeper.znode.rootserver		=	root-region-server

#hbase.zookeeper.quorum			=	namenode1.hadoop.dm.openaire.eu,namenode2.hadoop.dm.openaire.eu,jobtracker1.hadoop.dm.openaire.eu,jobtracker2.hadoop.dm.openaire.eu,hbase-master1.hadoop.dm.openaire.eu

#hbase.zookeeper.property.clientPort	=	2181

#hbase.zookeeper.client.port		=	2181

#zookeeper.znode.parent			=	/hbase

#

##HDFS-SITE

#dfs.replication					=	2

#dfs.nameservices				=	dm-cluster-nn

#dfs.ha.namenodes.dm-cluster-nn	=	nn1,nn2

#

#dfs.namenode.rpc-address.dm-cluster-nn.nn1=namenode1.hadoop.dm.openaire.eu:8020

#dfs.namenode.http-address.dm-cluster-nn.nn1=namenode1.hadoop.dm.openaire.eu:50070

#dfs.namenode.rpc-address.dm-cluster-nn.nn2=namenode2.hadoop.dm.openaire.eu:8020

#dfs.namenode.http-address.dm-cluster-nn.nn2=namenode2.hadoop.dm.openaire.eu:50070

#

#dfs.client.failover.proxy.provider.dm-cluster-nn=org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider

#OOZIE SERVER

oozie.service.loc =  http://hadoop-edge3.garr-pa1.d4science.org:11000/oozie/

oozie.service.loc				=	http://iis-cdh5-test-m1.ocean.icm.edu.pl:11000/oozie/

services.data.hadoop.garr.properties = classpath:/eu/dnetlib/data/hadoop/config/hadoop-default.dm.garr.properties

package eu.dnetlib.msro.workflows.nodes.dedup;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.workflows.graph.Arc;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.rmi.enabling.ISLookUpException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import org.apache.commons.lang3.StringUtils;

import org.springframework.beans.factory.annotation.Autowired;

import java.io.ByteArrayOutputStream;

import java.util.List;

import java.util.zip.GZIPOutputStream;

public class FindDedupConfigurationJobNode extends SimpleJobNode {

    private String configName;

    @Autowired

    UniqueServiceLocator uniqueServiceLocator;

    private final static String query ="for $x in collection('/db/DRIVER/DedupConfigurationDSResources/DedupConfigurationDSResourceType') where $x//RESOURCE_IDENTIFIER/@value ='%s' return $x//DEDUPLICATION/text()";

    @Override

    protected String execute(Env env) throws Exception {

        if (StringUtils.isBlank(configName)) throw new IllegalArgumentException("missing configuration sequence");

        env.setAttribute("dconf", getProfile());

        return Arc.DEFAULT_ARC;

    }

    private String getProfile() throws Exception {

        ISLookUpService service = uniqueServiceLocator.getService(ISLookUpService.class);

        List<String> resourceProfile = service.quickSearchProfile(String.format(query,configName));

        if (resourceProfile.size()!= 1) throw new IllegalStateException("Query should return on result query:"+String.format(query,configName));

        return  compressArgument(resourceProfile.get(0));

    }

    private static String compressArgument(final String value)  throws Exception{

        ByteArrayOutputStream out = new ByteArrayOutputStream();

        GZIPOutputStream gzip = new GZIPOutputStream(out);

        gzip.write(value.getBytes());

        gzip.close();

        return java.util.Base64.getEncoder().encodeToString(out.toByteArray());

    }

    public String getConfigName() {

        return configName;

    }

    public void setConfigName(String configName) {

        this.configName = configName;

    }

}

	<bean id="wfNodeFindDedupConfiguration" class="eu.dnetlib.msro.workflows.nodes.dedup.FindDedupConfigurationJobNode"

		  scope="prototype"/>

<RESOURCE_PROFILE>

	<HEADER>

		<RESOURCE_IDENTIFIER value="1dd10bf0-5c97-470c-9938-ae8e57a422fc_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl"/>

		<RESOURCE_TYPE value="WorkflowDSResourceType"/>

		<RESOURCE_KIND value="WorkflowDSResources"/>

		<RESOURCE_URI value=""/>

		<DATE_OF_CREATION value="2020-02-05T18:13:51.0Z"/>

	</HEADER>

	<BODY>

		<WORKFLOW_NAME menuSection="InfoSpace Deduplication">InfoSpace Deduplication using Spark</WORKFLOW_NAME>

		<WORKFLOW_DESCRIPTION>InfoSpace Deduplication using Spark</WORKFLOW_DESCRIPTION>

		<WORKFLOW_INFO/>

		<WORKFLOW_FAMILY>InfoSpace Deduplication</WORKFLOW_FAMILY>

		<WORKFLOW_PRIORITY>35</WORKFLOW_PRIORITY>

		<CONFIGURATION status="EXECUTABLE" start="MANUAL">

			<PARAMETERS>

				<PARAM description="Oozie Job name" function="listProfiles('HadoopJobConfigurationDSResourceType', '//HADOOP_JOB/@name','executeOozie')" managedBy="user" name="oozieJobName" required="true" type="string"/>

				<PARAM name="workingDirPath" description="working dir where generate all the intermediate verison of the graph" required="true" type="string" managedBy="user"/>

				<PARAM name="cluster" description="Hadoop cluster logical name" required="true" managedBy="user"   type="string" function="validValues(['DM','IIS', 'GARR'])"/>

				<PARAM name="reusePublication" description="reuse publications on HDFS?" required="true" type="boolean" managedBy="user"/>

				<PARAM name="reuseResolvedPublication" description="reuse resolved publications on HDFS?" required="true" type="boolean" managedBy="user"/>

				<PARAM name="reuseDataset" description="reuse datasets on HDFS?" required="true" type="boolean" managedBy="user"/>

				<PARAM name="reuseResolvedDataset" description="reuse resolved datasets on HDFS?" required="true" type="boolean" managedBy="user"/>

				<PARAM name="reuseUnresolved" description="reuse unresolved objects on HDFS?" required="true" type="boolean" managedBy="user"/>

                <PARAM name="dedupConfigDataset" description="dedup configuration orchestration name" required="true" type="string" function="obtainValues('dedupOrchestrations', {})" managedBy="user"/>

                <PARAM name="dedupConfigPublication" description="dedup configuration orchestration name" required="true" type="string" function="listProfiles('DedupConfigurationDSResources', '//DESCRIPTION', '')" managedBy="user"/>

				<PARAM name="dedupConfigUnknown" description="dedup configuration orchestration name" required="true" type="string" function="obtainValues('dedupOrchestrations', {})" managedBy="user"/>

			</PARAMETERS>

			<WORKFLOW>

				<NODE name="pmf2hdfs" type="LaunchWorkflowTemplate">

					<DESCRIPTION>import PMF Publications to HDFS DIR</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="wfTemplateId" value="4a268738-b635-4d86-9a4a-52bec6d20866_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

						<PARAM name="wfTemplateParams">

							<MAP>

								<ENTRY key="cluster" ref="cluster"/>

								<ENTRY key="reuseMdRecords" ref="reusePublication"/>

								<ENTRY key="mdFormat" value="PMF"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/xml/pmf.dli.seq"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="targetPath" value="${workingDirPath}/input/0"/>

								<ENTRY key="entity" value="publication"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="extractPublication"/>

					</ARCS>

				</NODE>

				<NODE name="extractPublication" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/extractentities/oozie_app"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/input/0"/>

								<ENTRY key="targetPath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetDir" value="0"/>

								<ENTRY key="entities" value="publication,unknown,relation"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="pmfResolved2hdfs"/>

					</ARCS>

				</NODE>

				<NODE name="pmfResolved2hdfs" type="LaunchWorkflowTemplate">

					<DESCRIPTION>import PMF Publications to HDFS DIR</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="wfTemplateId" value="4a268738-b635-4d86-9a4a-52bec6d20866_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

						<PARAM name="wfTemplateParams">

							<MAP>

								<ENTRY key="cluster" ref="cluster"/>

								<ENTRY key="reuseMdRecords" ref="reuseResolvedPublication"/>

								<ENTRY key="mdFormat" value="PMF"/>

								<ENTRY key="interpretation" value="resolved"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/xml/pmf.dli.resolved.seq"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="targetPath" value="${workingDirPath}/input/1"/>

								<ENTRY key="entity" value="publication"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="extractPublicationResolved"/>

					</ARCS>

				</NODE>

				<NODE name="extractPublicationResolved" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/extractentities/oozie_app"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/input/1"/>

								<ENTRY key="targetPath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetDir" value="1"/>

								<ENTRY key="entities" value="publication"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="dmf2hdfs"/>

					</ARCS>

				</NODE>

				<NODE name="dmf2hdfs" type="LaunchWorkflowTemplate">

					<DESCRIPTION>import PMF Publications to HDFS DIR</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="wfTemplateId" value="4a268738-b635-4d86-9a4a-52bec6d20866_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

						<PARAM name="wfTemplateParams">

							<MAP>

								<ENTRY key="cluster" ref="cluster"/>

								<ENTRY key="reuseMdRecords" ref="reuseDataset"/>

								<ENTRY key="mdFormat" value="DMF"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/xml/dmf.dli.seq"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="targetPath" value="${workingDirPath}/input/2"/>

								<ENTRY key="entity" value="dataset"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="extractDataset"/>

					</ARCS>

				</NODE>

				<NODE name="extractDataset" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/extractentities/oozie_app"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/input/2"/>

								<ENTRY key="targetPath" value="${workingDirPath}/extracted"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="targetDir" value="2"/>

								<ENTRY key="entities" value="dataset,unknown,relation"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="dmfResolved2hdfs"/>

					</ARCS>

				</NODE>

				<NODE name="dmfResolved2hdfs" type="LaunchWorkflowTemplate">

					<DESCRIPTION>import PMF Publications to HDFS DIR</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="wfTemplateId" value="4a268738-b635-4d86-9a4a-52bec6d20866_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

						<PARAM name="wfTemplateParams">

							<MAP>

								<ENTRY key="cluster" ref="cluster"/>

								<ENTRY key="reuseMdRecords" ref="reuseResolvedDataset"/>

								<ENTRY key="mdFormat" value="DMF"/>

								<ENTRY key="interpretation" value="resolved"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/xml/dmf.dli.resolved.seq"/>

								<ENTRY key="targetPath" value="${workingDirPath}/input/3"/>

								<ENTRY key="entity" value="dataset"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="extractDatasetResolved"/>

					</ARCS>

				</NODE>

				<NODE name="extractDatasetResolved" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/extractentities/oozie_app"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/input/3"/>

								<ENTRY key="targetPath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetDir" value="3"/>

								<ENTRY key="entities" value="dataset"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="mergeDataset"/>

					</ARCS>

				</NODE>

				<NODE isStart="true" name="mergeDataset" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/mergeentities/oozie_app"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetPath" value="${workingDirPath}/graph"/>

								<ENTRY key="entity" value="dataset"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="mergePublication"/>

					</ARCS>

				</NODE>

				<NODE name="mergePublication" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/mergeentities/oozie_app"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetPath" value="${workingDirPath}/graph"/>

								<ENTRY key="entity" value="publication"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="mergeUnknown"/>

					</ARCS>

				</NODE>

				<NODE name="mergeUnknown" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/mergeentities/oozie_app"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetPath" value="${workingDirPath}/graph"/>

								<ENTRY key="entity" value="unknown"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="mergeRelation"/>

					</ARCS>

				</NODE>

				<NODE name="mergeRelation" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJobName"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/mergeentities/oozie_app"/>

								<ENTRY key="workingDirPath" value="${workingDirPath}"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/extracted"/>

								<ENTRY key="targetPath" value="${workingDirPath}/graph"/>

								<ENTRY key="entity" value="relation"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="success"/>

					</ARCS>

				</NODE>

				<NODE name="dedupPublication" type="LaunchWorkflowTemplate">

					<DESCRIPTION>import PMF Publications to HDFS DIR</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="wfTemplateId" value="b8e1afcf-e5ca-47d0-9ee8-47da90e1a9c3_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

						<PARAM name="wfTemplateParams">

							<MAP>

								<ENTRY key="cluster" ref="cluster"/>

								<ENTRY key="sourcePath" value="${workingDirPath}/graph"/>

								<ENTRY key="targetPath" value="${workingDirPath}/dedupGraphWD"/>

								<ENTRY key="entity" value="publication"/>

								<ENTRY key="dedup_conf" ref="dedupConfigPublication"/>

								<ENTRY key="oozieJob" ref="oozieJobName"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="success"/>

					</ARCS>

				</NODE>

			</WORKFLOW>

		</CONFIGURATION>

		<NOTIFICATIONS/>

		<SCHEDULING enabled="false">

			<CRON>29 5 22 ? * *</CRON>

			<MININTERVAL>10080</MININTERVAL>

		</SCHEDULING>

		<STATUS/>

	</BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="b8e1afcf-e5ca-47d0-9ee8-47da90e1a9c3_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="WorkflowTemplateDSResourceType"/>

        <RESOURCE_KIND value="WorkflowTemplateDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2020-02-14T12:27:20+01:00"/>

    </HEADER>

    <BODY>

        <CONFIGURATION>

            <PARAMETERS>

                <PARAM description="HDFS path" name="sourcePath" required="true" type="string"/>

                <PARAM description="HDFS path" name="targetPath" required="true" type="string"/>

                <PARAM description="HDFS path" name="entity" required="true" type="string"/>

                <PARAM description="Hadoop cluster name" name="cluster" required="true" type="string"/>

                <PARAM description="Dedup Configuration" name="dedup_conf" required="true" type="string"/>

            </PARAMETERS>

            <WORKFLOW>

                <NODE isStart="true" name="findDedupConfiguration" type="FindDedupConfiguration">

                    <DESCRIPTION>Find Dedup configuration</DESCRIPTION>

                    <PARAMETERS>

                        <PARAM name="configName" ref="dedup_conf"/>

                    </PARAMETERS>

                    <ARCS>

                        <ARC to="dedupPublication"/>

                    </ARCS>

                </NODE>

                <NODE name="dedupPublication" type="SubmitHadoopJob">

                    <DESCRIPTION>Run M/R import Job</DESCRIPTION>

                    <PARAMETERS>

                        <PARAM name="cluster" ref="cluster"/>

                        <PARAM name="hadoopJob" value="executeOozieJobGARR"/>

                        <PARAM name="jobParams">

                            <MAP>

                                <ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/dedup/oozie_app"/>

                                <ENTRY key="sourcePath" ref="sourcePath"/>

                                <ENTRY key="targetPath"  ref="targetPath"/>

                                <ENTRY key="entity" ref="entity"/>

                                <ENTRY key="dedupConf" env="dconf"/>

                            </MAP>

                        </PARAM>

                    </PARAMETERS>

                    <ARCS>

                        <ARC to="success"/>

                    </ARCS>

                </NODE>

            </WORKFLOW>

        </CONFIGURATION>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="6f111858-1719-48a9-8e90-8f829497f7a8_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="WorkflowTemplateDSResourceType"/>

        <RESOURCE_KIND value="WorkflowTemplateDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2016-06-20T07:52:08+00:00"/>

    </HEADER>

    <BODY>

        <CONFIGURATION>

            <PARAMETERS>

                <PARAM name="targetDir" description="metadata interpretation name" required="false" type="string" default="cleaned"/>

                <PARAM name="sourcePath" description="HDFS path" required="true" type="string"/>

                <PARAM name="targetPath" description="HDFS path" required="true" type="string"/>

                <PARAM name="entity" description="HDFS path" required="true" type="string"/>

                <PARAM name="cluster" description="Hadoop cluster name" required="true" type="string"/>

            </PARAMETERS>

            <WORKFLOW>

                <NODE name="convertEntity" type="SubmitHadoopJob">

                    <DESCRIPTION>Run M/R import Job</DESCRIPTION>

                    <PARAMETERS>

                        <PARAM name="cluster" ref="cluster"/>

                        <PARAM name="hadoopJob" value="importMetadataToHDFS"/>

                        <PARAM name="jobParams">

                            <MAP>

                                <ENTRY key="sourcePath" ref="sourcePath"/>

                                <ENTRY key="targetPath" ref="targetPath"/>

                                <ENTRY key="entity" ref="entity"/>

                            </MAP>

                        </PARAM>

                    </PARAMETERS>

                    <ARCS>

                        <ARC to="success"/>

                    </ARCS>

                </NODE>

            </WORKFLOW>

        </CONFIGURATION>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="ea5ea1a0-a750-42db-8dfa-1606158698da_SGFkb29wSm9iQ29uZmlndXJhdGlvbkRTUmVzb3VyY2VzL0hhZG9vcEpvYkNvbmZpZ3VyYXRpb25EU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="HadoopJobConfigurationDSResourceType"/>

        <RESOURCE_KIND value="HadoopJobConfigurationDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2019-04-12T13:16:20+02:00"/>

    </HEADER>

    <BODY>

        <HADOOP_JOB name="importMetadataToHDFS" type="oozie">

            <DESCRIPTION>Import XML Data into sequence File</DESCRIPTION>

            <STATIC_CONFIGURATION><!-- Cluster wide -->

                <PROPERTY key="queueName" value="default"/>

                <PROPERTY key="user.name" value="sandro.labruzzo"/><!-- Runtime -->

                <PROPERTY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/oozie_app/"/>

                <PROPERTY key="oozie.wf.validate.ForkJoin" value="false"/>

                <PROPERTY key="oozie.use.system.libpath" value="True"/>

                <PROPERTY key="security_enabled" value="False"/>

                <PROPERTY key="dryrun" value="True"/>

                <PROPERTY key="oozie.action.sharelib.for.spark" value="spark2"/>

                <PROPERTY key="sparkDriverMemory" value="3G"/>

                <PROPERTY key="sparkExecutorMemory" value="3G"/>

                <PROPERTY key="metadataEncoding" value="XML"/>

            </STATIC_CONFIGURATION>

            <JOB_INTERFACE>

                <PARAM description="the path of the input hdfs file contains xml" name="sourcePath" required="true"/>

                <PARAM description="the path of the result hdfs containig OAF entities" name="targetPath" required="true"/>

                <PARAM description="The entity type" name="entity" required="true"/>

            </JOB_INTERFACE>

        </HADOOP_JOB>

        <STATUS>

            <LAST_SUBMISSION_DATE value="2019-04-17T17:46:31+02:00"/>

            <RUNNING_INSTANCES value="2"/>

            <CUMULATIVE_RUN value="83"/>

        </STATUS>

        <SECURITY_PARAMETERS>SECURITY_PARAMETERS</SECURITY_PARAMETERS>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="80b642da-533c-4c6f-b896-fbba12146175_SGFkb29wSm9iQ29uZmlndXJhdGlvbkRTUmVzb3VyY2VzL0hhZG9vcEpvYkNvbmZpZ3VyYXRpb25EU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="HadoopJobConfigurationDSResourceType"/>

        <RESOURCE_KIND value="HadoopJobConfigurationDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2019-04-12T13:16:20+02:00"/>

    </HEADER>

    <BODY>

        <HADOOP_JOB name="mergeEntitiesToHDFS" type="oozie">

            <DESCRIPTION>Import XML Data into sequence File</DESCRIPTION>

            <STATIC_CONFIGURATION><!-- Cluster wide -->

                <PROPERTY key="queueName" value="default"/>

                <PROPERTY key="user.name" value="sandro.labruzzo"/><!-- Runtime -->

                <PROPERTY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/mergeentities/oozie_app/"/>

                <PROPERTY key="oozie.wf.validate.ForkJoin" value="false"/>

                <PROPERTY key="oozie.use.system.libpath" value="True"/>

                <PROPERTY key="security_enabled" value="False"/>

                <PROPERTY key="dryrun" value="True"/>

                <PROPERTY key="oozie.action.sharelib.for.spark" value="spark2"/>

                <PROPERTY key="sparkDriverMemory" value="4G"/>

                <PROPERTY key="sparkExecutorMemory" value="4G"/>

                <PROPERTY key="metadataEncoding" value="XML"/>

            </STATIC_CONFIGURATION>

            <JOB_INTERFACE>

                <PARAM description="the path of the input hdfs file contains extracted Entities" name="sourcePath" required="true"/>

                <PARAM description="the baseDir path of the result hdfs containig OAF merged entities" name="targetPath" required="true"/>

                <PARAM description="The entity type" name="entity" required="true"/>

            </JOB_INTERFACE>

        </HADOOP_JOB>

        <STATUS>

            <LAST_SUBMISSION_DATE value="2019-04-17T17:46:31+02:00"/>

            <RUNNING_INSTANCES value="2"/>

            <CUMULATIVE_RUN value="83"/>

        </STATUS>

        <SECURITY_PARAMETERS>SECURITY_PARAMETERS</SECURITY_PARAMETERS>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="ed671880-d053-4692-90e4-5901ddde661b_SGFkb29wSm9iQ29uZmlndXJhdGlvbkRTUmVzb3VyY2VzL0hhZG9vcEpvYkNvbmZpZ3VyYXRpb25EU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="HadoopJobConfigurationDSResourceType"/>

        <RESOURCE_KIND value="HadoopJobConfigurationDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2020-02-17T15:36:24+01:00"/>

    </HEADER>

    <BODY>

        <HADOOP_JOB name="executeOozieJobICM" type="oozie">

            <DESCRIPTION>Import XML Data into sequence File</DESCRIPTION>

            <STATIC_CONFIGURATION><!-- Cluster wide -->

                <PROPERTY key="queueName" value="default"/>

                <PROPERTY key="user.name" value="sandro.labruzzo"/>

                <PROPERTY key="oozie.wf.validate.ForkJoin" value="false"/>

                <PROPERTY key="oozie.use.system.libpath" value="True"/>

                <PROPERTY key="security_enabled" value="False"/>

                <PROPERTY key="dryrun" value="True"/>

                <PROPERTY key="oozie.action.sharelib.for.spark" value="spark2"/>

                <PROPERTY key="sparkDriverMemory" value="4G"/>

                <PROPERTY key="sparkExecutorMemory" value="4G"/>

                <PROPERTY key="metadataEncoding" value="XML"/>

                <PROPERTY key="jobTracker" value="yarnRM"/>

                <PROPERTY key="nameNode" value="hdfs://nameservice1"/>

                <PROPERTY key="sparkExtraOPT" value="--conf spark.extraListeners=&#34;com.cloudera.spark.lineage.NavigatorAppListener&#34; --conf spark.sql.queryExecutionListeners=&#34;com.cloudera.spark.lineage.NavigatorQueryListener&#34; --conf spark.sql.warehouse.dir=&#34;/user/hive/warehouse&#34;"/>

                <PROPERTY key="projectVersion" value="1.0.5-SNAPSHOT"/>

            </STATIC_CONFIGURATION>

            <JOB_INTERFACE>

                <PARAM description="the path of the input hdfs file contains xml" name="oozie.wf.application.path" required="true"/>

            </JOB_INTERFACE>

        </HADOOP_JOB>

        <STATUS>

            <LAST_SUBMISSION_DATE value="2020-02-17T14:18:37+01:00"/>

            <RUNNING_INSTANCES value="4"/>

            <CUMULATIVE_RUN value="105"/>

        </STATUS>

        <SECURITY_PARAMETERS>SECURITY_PARAMETERS</SECURITY_PARAMETERS>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

    <HEADER>

        <RESOURCE_IDENTIFIER value="c277d01e-d10b-4150-8d9f-caf524e40fe3_SGFkb29wSm9iQ29uZmlndXJhdGlvbkRTUmVzb3VyY2VzL0hhZG9vcEpvYkNvbmZpZ3VyYXRpb25EU1Jlc291cmNlVHlwZQ=="/>

        <RESOURCE_TYPE value="HadoopJobConfigurationDSResourceType"/>

        <RESOURCE_KIND value="HadoopJobConfigurationDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2019-04-12T13:16:20+02:00"/>

    </HEADER>

    <BODY>

        <HADOOP_JOB name="executeOozieJobGARR" type="oozie">

            <DESCRIPTION>Import XML Data into sequence File</DESCRIPTION>

            <STATIC_CONFIGURATION><!-- Cluster wide -->

                <PROPERTY key="queueName" value="default"/>

                <PROPERTY key="user.name" value="sandro.labruzzo"/><!-- Runtime -->

<!--                <PROPERTY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/oozie_app/"/>-->

                <PROPERTY key="oozie.wf.validate.ForkJoin" value="false"/>

                <PROPERTY key="oozie.use.system.libpath" value="True"/>

                <PROPERTY key="security_enabled" value="False"/>

                <PROPERTY key="dryrun" value="True"/>

                <PROPERTY key="oozie.action.sharelib.for.spark" value="spark2"/>

                <PROPERTY key="sparkDriverMemory" value="4G"/>

                <PROPERTY key="sparkExecutorMemory" value="4G"/>

                <PROPERTY key="metadataEncoding" value="XML"/>

                <PROPERTY key="jobTracker" value="hadoop-rm3.garr-pa1.d4science.org:8032"/>

                <PROPERTY key="nameNode" value="hdfs://hadoop-rm1.garr-pa1.d4science.org:8020"/>

                <PROPERTY key="projectVersion" value="1.0.5-SNAPSHOT"/>

            </STATIC_CONFIGURATION>

            <JOB_INTERFACE>

                <PARAM description="the path of the input hdfs file contains xml" name="oozie.wf.application.path" required="true"/>

            </JOB_INTERFACE>

        </HADOOP_JOB>

        <STATUS>

            <LAST_SUBMISSION_DATE value="2019-04-17T17:46:31+02:00"/>

            <RUNNING_INSTANCES value="2"/>

            <CUMULATIVE_RUN value="83"/>

        </STATUS>

        <SECURITY_PARAMETERS>SECURITY_PARAMETERS</SECURITY_PARAMETERS>

    </BODY>

</RESOURCE_PROFILE>

<RESOURCE_PROFILE>

	<HEADER>

		<RESOURCE_IDENTIFIER value="4a268738-b635-4d86-9a4a-52bec6d20866_V29ya2Zsb3dUZW1wbGF0ZURTUmVzb3VyY2VzL1dvcmtmbG93VGVtcGxhdGVEU1Jlc291cmNlVHlwZQ=="/>

		<RESOURCE_TYPE value="WorkflowTemplateDSResourceType"/>

		<RESOURCE_KIND value="WorkflowTemplateDSResources"/>

		<RESOURCE_URI value=""/>

		<DATE_OF_CREATION value="2020-02-17T16:16:24+01:00"/>

	</HEADER>

	<BODY>

		<CONFIGURATION>

			<PARAMETERS>

				<PARAM description="reuse metadata records?" name="reuseMdRecords" required="true" type="boolean"/>

				<PARAM description="metadata format name" name="mdFormat" required="true" type="string"/>

				<PARAM default="cleaned" description="metadata interpretation name" name="interpretation" required="false" type="string"/>

				<PARAM description="HDFS path" name="sourcePath" required="true" type="string"/>

				<PARAM description="HDFS path" name="targetPath" required="true" type="string"/>

				<PARAM description="HDFS path" name="entity" required="true" type="string"/>

				<PARAM description="Hadoop cluster name" name="cluster" required="true" type="string"/>

				<PARAM description="Oozie Job Name" name="oozieJob" required="true" type="string"/>

			</PARAMETERS>

			<WORKFLOW>

				<NODE isStart="true" name="reuseHdfsRecords" type="ReuseHdfsRecords">

					<DESCRIPTION>reuse mdstore records</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="reuseMdRecords" ref="reuseMdRecords"/>

					</PARAMETERS>

					<ARCS>

						<ARC name="true" to="doneExport"/>

						<ARC name="false" to="exportRecords"/>

					</ARCS>

				</NODE>

				<NODE name="exportRecords" type="MDStoreBatchExporter">

					<DESCRIPTION>Fetch mdstore records</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="format" ref="mdFormat"/>

						<PARAM name="layout" value="store"/>

						<PARAM name="interpretation" ref="interpretation"/>

						<PARAM name="outputEprParam" value="records_epr"/>

					</PARAMETERS>

					<ARCS>

						<ARC to="storeHdfsRecords"/>

					</ARCS>

				</NODE>

				<NODE name="storeHdfsRecords" type="StoreHdfsRecords">

					<DESCRIPTION>Store records to HDFS</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="inputEprParam" value="records_epr"/>

						<PARAM name="hdfsPath" ref="hdfsPath"/>

						<PARAM name="cluster" ref="cluster"/>

					</PARAMETERS>

					<ARCS>

						<ARC to="doneExport"/>

					</ARCS>

				</NODE>

				<NODE name="doneExport">

					<DESCRIPTION/>

					<PARAMETERS/>

					<ARCS>

						<ARC to="convertEntity"/>

					</ARCS>

				</NODE>

				<NODE name="convertEntity" type="SubmitHadoopJob">

					<DESCRIPTION>Run M/R import Job</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="cluster" ref="cluster"/>

						<PARAM name="hadoopJob" ref="oozieJob"/>

						<PARAM name="jobParams">

							<MAP>

								<ENTRY key="sourcePath" ref="sourcePath"/>

								<ENTRY key="targetPath" ref="targetPath"/>

								<ENTRY key="entity" ref="entity"/>

								<ENTRY key="oozie.wf.application.path" value="/user/sandro.labruzzo/graph/scholexplorer/oozie_app"/>

							</MAP>

						</PARAM>

					</PARAMETERS>

					<ARCS>

						<ARC to="success"/>

					</ARCS>

				</NODE>

			</WORKFLOW>

		</CONFIGURATION>

	</BODY>

</RESOURCE_PROFILE>

	GARR, // Data Management at GARR with spark2

package eu.dnetlib.graph;

import org.junit.Test;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

public class GraphLoaderTest {

    final String regex = "\\$\\{(\\w*)\\}";

    final Pattern pattern = Pattern.compile(regex, Pattern.MULTILINE);

    @Test

    public void testRegEx () {

        String string = "${1234}/sdfjasdfpojawpdf/${dd}";

        final Matcher matcher = pattern.matcher(string);

        int k = 0;

        while (matcher.find()) {

            System.out.println("Full match: " + matcher.group(0));

            for (int i = 1; i <= matcher.groupCount(); i++) {

                System.out.println("Group " + i + ": " + matcher.group(i));

                string = string.replaceAll(Pattern.quote(matcher.group(0)), "VALORE"+k++);

                System.out.println("new String  = " + string);

            }

        }

    }

}

import java.util.regex.Matcher;

import java.util.regex.Pattern;

    private static final Log log = LogFactory.getLog(GraphLoader.class);

    private NodeHelper nodeHelper;

    private String regExRef = "\\$\\{(\\w*)\\}";

    final Pattern pattern = Pattern.compile(regExRef, Pattern.MULTILINE);

    @Resource(name = "propertyFetcher")

    private PropertyFetcher propertyFetcher;

    public Graph loadGraph(final Document doc, final Map<String, String> globalParams) throws MSROException {

        final Graph graph = new Graph();

        for (final Object o : doc.selectNodes("//CONFIGURATION/WORKFLOW/NODE")) {

            final Element n = (Element) o;

            final String nodeName = n.valueOf("@name");

            final String nodeType = n.valueOf("@type");

            final boolean isStart = StringUtils.equalsIgnoreCase(n.valueOf("@isStart"), "true");

            final boolean isJoin = StringUtils.equalsIgnoreCase(n.valueOf("@isJoin"), "true");

            final Map<String, GraphNodeParameter> params = calculateParamsForNode(n, globalParams);

            if (isStart) {

                graph.addNode(GraphNode.newStartNode(nodeName, nodeType, params));

            } else if (isJoin) {

                graph.addNode(GraphNode.newJoinNode(nodeName, nodeType, params));

            } else {

                graph.addNode(GraphNode.newNode(nodeName, nodeType, params));

            }

            for (final Object o1 : n.selectNodes(".//ARC")) {

                final Element a = (Element) o1;

                final String arcName = a.valueOf("@name");

                final String to = a.valueOf("@to");

                graph.addArc(new Arc(StringUtils.isNotBlank(arcName) ? arcName : Arc.DEFAULT_ARC, nodeName, to));

            }

            graph.addNode(GraphNode.newSuccessNode());

        }

        checkValidity(graph);

        return graph;

    }

    public Map<String, GraphNodeParameter> calculateParamsForNode(final Node node, final Map<String, String> globalParams) {

        final Map<String, GraphNodeParameter> params = new HashMap<>();

        if (node != null) {

            for (final Object o : node.selectNodes(".//PARAM")) {

                final Element p = (Element) o;

                final String pName = p.valueOf("@name");

                final GraphNodeParameter pValue = calculateSimpleValue((Element) o, globalParams);

                if (pValue != null) {

                    params.put(pName, pValue);

                } else if (p.selectSingleNode("./MAP") != null) {

                    @SuppressWarnings("unchecked") final Map<String, GraphNodeParameter> map = ((List<Element>) p.selectNodes("./MAP/ENTRY"))

                            .stream()

                            .collect(Collectors.toMap(

                                    e -> e.valueOf("@key"),

                                    e -> {

                                        final GraphNodeParameter gnp = calculateSimpleValue(e, globalParams);

                                        if (gnp == null) {

                                            final String msg = String.format("missing value for param: \"%s\"", e.valueOf("@key"));

                                            log.debug(msg);

                                            return GraphNodeParameter.newNullParam();

                                        }

                                        return gnp;

                                    }));

                    params.put(pName, GraphNodeParameter.newMapParam(map));

                } else if (p.selectSingleNode("./LIST") != null) {

                    @SuppressWarnings("unchecked") final List<GraphNodeParameter> list = ((List<Element>) p.selectNodes("./LIST/ITEM"))

                            .stream()

                            .map(e -> calculateSimpleValue(e, globalParams))

                            .collect(Collectors.toList());

                    params.put(pName, GraphNodeParameter.newListParam(list));

                }

            }

        }

        return params;

    }

    private GraphNodeParameter calculateSimpleValue(final Element elem, final Map<String, String> globalParams) {

        String value = elem.valueOf("@value");

        final String ref = elem.valueOf("@ref");

        final String prop = elem.valueOf("@property");

        final String envRef = elem.valueOf("@env");

        if (StringUtils.isNotBlank(ref) && StringUtils.isNotBlank(globalParams.get(ref))) {

            return GraphNodeParameter.newSimpleParam(globalParams.get(ref));

        } else if (StringUtils.isNotBlank(envRef)) {

            return GraphNodeParameter.newEnvParam(envRef);

        } else if (StringUtils.isNotBlank(value)) {

            Matcher matcher = pattern.matcher(value);

            while (matcher.find()) {

                final String rName = matcher.group(1);

                final String rValue = globalParams.get(rName);

                if (StringUtils.isBlank(rValue)) {

                	return null;

				}

                value = value.replaceAll(Pattern.quote(matcher.group(0)), rValue);

				System.out.println("NEW VALUE "+value);

            }

            return GraphNodeParameter.newSimpleParam(value);

        } else if (StringUtils.isNotBlank(prop)) {

            return GraphNodeParameter.newSimpleParam(this.propertyFetcher.getProperty(prop));

        } else {

            return null;

        }

    }

    private void checkValidity(final Graph graph) throws MSROException {

        final Set<String> nodesFromArcs = new HashSet<String>();

        boolean foundSuccess = false;

        boolean foundStart = false;

        for (final Arc arc : graph.getArcs()) {

            if (StringUtils.isBlank(arc.getFrom()) || StringUtils.isBlank(arc.getFrom())) {

                throw new MSROException("Invalid arc: missing from e/o to");

            }

            if (StringUtils.equals(arc.getTo(), GraphNode.SUCCESS_NODE)) {

                foundSuccess = true;

            }

            nodesFromArcs.add(arc.getFrom());

            nodesFromArcs.add(arc.getTo());

        }

        if (!foundSuccess) {

            throw new MSROException("Arc to success not found");

        }

        final Set<String> diff = Sets.symmetricDifference(graph.nodeNames(), nodesFromArcs);

        if (!diff.isEmpty()) {

            throw new MSROException("Missing or invalid nodes in arcs: " + diff);

        }

        for (final GraphNode n : graph.nodes()) {

            if (StringUtils.isBlank(n.getName())) {

                throw new MSROException("Invalid node: missing name");

            }

            if (n.isStart()) {

                foundStart = true;

            }

            if (!this.nodeHelper.isValidType(n.getType())) {

                throw new MSROException("Invalid node type: " + n.getType());

            }

        }

        if (!foundStart) {

            throw new MSROException("Start node not found");

        }

    }

    public NodeHelper getNodeHelper() {

        return this.nodeHelper;

    }

    @Required

    public void setNodeHelper(final NodeHelper nodeHelper) {

        this.nodeHelper = nodeHelper;

    }