Merge pull request 'enrichmentSingleStep' (#373) from enrichmentSingleStep into beta

Reviewed-on: D-Net/dnet-hadoop#373

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/api/Utils.java

@@ -167,4 +167,11 @@ public class Utils implements Serializable {
 			});
 		return projectMap;
 	}
+
+	public static List<String> getCommunityIdList(String baseURL) throws IOException {
+		return getValidCommunities(baseURL)
+			.stream()
+			.map(community -> community.getId())
+			.collect(Collectors.toList());
+	}
 }

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/bulktag/SparkBulkTagJob.java

@@ -45,7 +45,7 @@ public class SparkBulkTagJob {
 			.toString(
 				SparkBulkTagJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/bulktag/input_bulkTag_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/bulktag/input_bulkTag_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 		parser.parseArgument(args);
@@ -105,7 +105,6 @@ public class SparkBulkTagJob {
 		Map<String, List<Pair<String, SelectionConstraints>>> dsm = cc.getEoscDatasourceMap();
 
 		for (String ds : datasources.collectAsList()) {
-			// final String dsId = ds.substring(3);
 			if (!dsm.containsKey(ds)) {
 				ArrayList<Pair<String, SelectionConstraints>> eoscList = new ArrayList<>();
 				dsm.put(ds, eoscList);
@@ -116,13 +115,11 @@ public class SparkBulkTagJob {
 
 	private static boolean isOKDatasource(Datasource ds) {
 		final String compatibility = ds.getOpenairecompatibility().getClassid();
-		boolean isOk = (compatibility.equalsIgnoreCase(OPENAIRE_3) ||
+		return (compatibility.equalsIgnoreCase(OPENAIRE_3) ||
 			compatibility.equalsIgnoreCase(OPENAIRE_4) ||
 			compatibility.equalsIgnoreCase(OPENAIRE_CRIS) ||
 			compatibility.equalsIgnoreCase(OPENAIRE_DATA)) &&
 			ds.getCollectedfrom().stream().anyMatch(cf -> cf.getKey().equals(EOSC));
-
-		return isOk;
 	}
 
 	private static <R extends Result> void execBulkTag(
@@ -151,7 +148,13 @@ public class SparkBulkTagJob {
 					.write()
 					.mode(SaveMode.Overwrite)
 					.option("compression", "gzip")
-					.json(outputPath + e.name());
+					.json(outputPath + e.name());// writing the tagging in the working dir for entity
+
+				readPath(spark, outputPath + e.name(), resultClazz) // copy the tagging in the actual result output path
+					.write()
+					.mode(SaveMode.Overwrite)
+					.option("compression", "gzip")
+					.json(inputPath + e.name());
 			});
 
 	}

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java

@@ -45,7 +45,7 @@ public class PrepareDatasourceCountryAssociation {
 			.toString(
 				PrepareDatasourceCountryAssociation.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/countrypropagation/input_prepareassoc_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/countrypropagation/input_prepareassoc_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -66,7 +66,7 @@ public class PrepareDatasourceCountryAssociation {
 			conf,
 			isSparkSessionManaged,
 			spark -> {
-				removeOutputDir(spark, outputPath);
+				// removeOutputDir(spark, outputPath);
 				prepareDatasourceCountryAssociation(
 					spark,
 					Arrays.asList(parser.get("whitelist").split(";")),
@@ -90,7 +90,8 @@ public class PrepareDatasourceCountryAssociation {
 				(FilterFunction<Datasource>) ds -> !ds.getDataInfo().getDeletedbyinference() &&
 					Optional.ofNullable(ds.getDatasourcetype()).isPresent() &&
 					Optional.ofNullable(ds.getDatasourcetype().getClassid()).isPresent() &&
-					(allowedtypes.contains(ds.getDatasourcetype().getClassid()) ||
+					((Optional.ofNullable(ds.getJurisdiction()).isPresent() &&
+						allowedtypes.contains(ds.getJurisdiction().getClassid())) ||
 						whitelist.contains(ds.getId())));
 
 		// filtering of the relations taking the non deleted by inference and those with IsProvidedBy as relclass

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareResultCountrySet.java

@@ -32,7 +32,7 @@ public class PrepareResultCountrySet {
 			.toString(
 				PrepareResultCountrySet.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/countrypropagation/input_prepareresultcountry_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/countrypropagation/input_prepareresultcountry_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java

@@ -35,7 +35,7 @@ public class SparkCountryPropagationJob {
 			.toString(
 				SparkCountryPropagationJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/countrypropagation/input_countrypropagation_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/countrypropagation/input_countrypropagation_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -97,6 +97,12 @@ public class SparkCountryPropagationJob {
 			.mode(SaveMode.Overwrite)
 			.json(outputPath);
 
+		readPath(spark, outputPath, resultClazz)
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(sourcePath);
+
 	}
 
 	private static <R extends Result> MapFunction<Tuple2<R, ResultCountrySet>, R> getCountryMergeFn() {

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/entitytoorganizationfromsemrel/PrepareInfo.java

@@ -60,7 +60,7 @@ public class PrepareInfo implements Serializable {
 			.toString(
 				SparkResultToOrganizationFromIstRepoJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/entitytoorganizationfromsemrel/input_preparation_parameter.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/entitytoorganizationfromsemrel/input_preparation_parameter.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/entitytoorganizationfromsemrel/SparkEntityToOrganizationFromSemRel.java

@@ -27,8 +27,8 @@ import eu.dnetlib.dhp.resulttoorganizationfrominstrepo.SparkResultToOrganization
 import eu.dnetlib.dhp.schema.common.ModelConstants;
 import eu.dnetlib.dhp.schema.oaf.Relation;
 
-public class SparkResultToOrganizationFromSemRel implements Serializable {
+public class SparkEntityToOrganizationFromSemRel implements Serializable {
-	private static final Logger log = LoggerFactory.getLogger(SparkResultToOrganizationFromSemRel.class);
+	private static final Logger log = LoggerFactory.getLogger(SparkEntityToOrganizationFromSemRel.class);
 	private static final int MAX_ITERATION = 5;
 	public static final String NEW_RESULT_RELATION_PATH = "/newResultRelation";
 	public static final String NEW_PROJECT_RELATION_PATH = "/newProjectRelation";
@@ -39,7 +39,7 @@ public class SparkResultToOrganizationFromSemRel implements Serializable {
 			.toString(
 				SparkResultToOrganizationFromIstRepoJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/entitytoorganizationfromsemrel/input_propagation_parameter.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/entitytoorganizationfromsemrel/input_propagation_parameter.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/entitytoorganizationfromsemrel/StepActions.java

@@ -3,8 +3,8 @@ package eu.dnetlib.dhp.entitytoorganizationfromsemrel;
 
 import static eu.dnetlib.dhp.PropagationConstant.*;
 import static eu.dnetlib.dhp.PropagationConstant.readPath;
-import static eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkResultToOrganizationFromSemRel.NEW_PROJECT_RELATION_PATH;
+import static eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkEntityToOrganizationFromSemRel.NEW_PROJECT_RELATION_PATH;
-import static eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkResultToOrganizationFromSemRel.NEW_RESULT_RELATION_PATH;
+import static eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkEntityToOrganizationFromSemRel.NEW_RESULT_RELATION_PATH;
 
 import java.io.Serializable;
 import java.util.*;
@@ -20,7 +20,6 @@ import org.jetbrains.annotations.NotNull;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import eu.dnetlib.dhp.KeyValueSet;
-import eu.dnetlib.dhp.schema.common.ModelConstants;
 import eu.dnetlib.dhp.schema.oaf.Relation;
 import scala.Tuple2;
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/PrepareResultOrcidAssociationStep1.java

@@ -31,7 +31,7 @@ public class PrepareResultOrcidAssociationStep1 {
 			.toString(
 				PrepareResultOrcidAssociationStep1.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_prepareorcidtoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/input_prepareorcidtoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConf);
 		parser.parseArgument(args);

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/PrepareResultOrcidAssociationStep2.java

@@ -29,7 +29,7 @@ public class PrepareResultOrcidAssociationStep2 {
 			.toString(
 				PrepareResultOrcidAssociationStep2.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_prepareorcidtoresult_parameters2.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/input_prepareorcidtoresult_parameters2.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/orcidtoresultfromsemrel/SparkOrcidToResultFromSemRelJob.java

@@ -2,7 +2,7 @@
 package eu.dnetlib.dhp.orcidtoresultfromsemrel;
 
 import static eu.dnetlib.dhp.PropagationConstant.*;
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 
 import java.util.List;
 import java.util.Optional;
@@ -36,7 +36,7 @@ public class SparkOrcidToResultFromSemRelJob {
 			.toString(
 				SparkOrcidToResultFromSemRelJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/orcidtoresultfromsemrel/input_orcidtoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/input_orcidtoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 		parser.parseArgument(args);
@@ -65,9 +65,8 @@ public class SparkOrcidToResultFromSemRelJob {
 		Class<? extends Result> resultClazz = (Class<? extends Result>) Class.forName(resultClassName);
 
 		SparkConf conf = new SparkConf();
-		conf.set("hive.metastore.uris", parser.get("hive_metastore_uris"));
 
-		runWithSparkHiveSession(
+		runWithSparkSession(
 			conf,
 			isSparkSessionManaged,
 			spark -> {

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/projecttoresult/PrepareProjectResultsAssociation.java

@@ -28,7 +28,7 @@ public class PrepareProjectResultsAssociation {
 			.toString(
 				PrepareProjectResultsAssociation.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/projecttoresult/input_prepareprojecttoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/projecttoresult/input_prepareprojecttoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/projecttoresult/SparkResultToProjectThroughSemRelJob.java

@@ -33,7 +33,7 @@ public class SparkResultToProjectThroughSemRelJob {
 			.toString(
 				SparkResultToProjectThroughSemRelJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/projecttoresult/input_projecttoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/projecttoresult/input_projecttoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromorganization/PrepareResultCommunitySet.java

@@ -34,7 +34,7 @@ public class PrepareResultCommunitySet {
 			.toString(
 				PrepareResultCommunitySet.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromorganization/input_preparecommunitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromorganization/input_preparecommunitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 		parser.parseArgument(args);

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromorganization/SparkResultToCommunityFromOrganizationJob.java

@@ -36,7 +36,7 @@ public class SparkResultToCommunityFromOrganizationJob {
 			.toString(
 				SparkResultToCommunityFromOrganizationJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromorganization/input_communitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromorganization/input_communitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -92,6 +92,12 @@ public class SparkResultToCommunityFromOrganizationJob {
 						.mode(SaveMode.Overwrite)
 						.option("compression", "gzip")
 						.json(outputPath + e.name());
+
+					readPath(spark, outputPath + e.name(), resultClazz)
+						.write()
+						.mode(SaveMode.Overwrite)
+						.option("compression", "gzip")
+						.json(inputPath + e.name());
 				}
 			});
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromproject/PrepareResultCommunitySet.java

@@ -38,7 +38,7 @@ public class PrepareResultCommunitySet {
 			.toString(
 				PrepareResultCommunitySet.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromproject/input_preparecommunitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromproject/input_preparecommunitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 		parser.parseArgument(args);

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromproject/SparkResultToCommunityFromProject.java

@@ -44,7 +44,7 @@ public class SparkResultToCommunityFromProject implements Serializable {
 			.toString(
 				SparkResultToCommunityFromProject.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromproject/input_communitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromproject/input_communitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -102,6 +102,12 @@ public class SparkResultToCommunityFromProject implements Serializable {
 						.mode(SaveMode.Overwrite)
 						.option("compression", "gzip")
 						.json(outputPath + e.name());
+
+					readPath(spark, outputPath + e.name(), resultClazz)
+						.write()
+						.mode(SaveMode.Overwrite)
+						.option("compression", "gzip")
+						.json(inputPath + e.name());
 				}
 			});
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromsemrel/PrepareResultCommunitySetStep1.java

@@ -4,6 +4,7 @@ package eu.dnetlib.dhp.resulttocommunityfromsemrel;
 import static eu.dnetlib.dhp.PropagationConstant.*;
 import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkHiveSession;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 
@@ -15,6 +16,7 @@ import org.slf4j.LoggerFactory;
 
 import com.google.gson.Gson;
 
+import eu.dnetlib.dhp.api.Utils;
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.resulttocommunityfromorganization.ResultCommunityList;
 import eu.dnetlib.dhp.schema.oaf.Relation;
@@ -26,11 +28,6 @@ import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
 public class PrepareResultCommunitySetStep1 {
 	private static final Logger log = LoggerFactory.getLogger(PrepareResultCommunitySetStep1.class);
 
-	private static final String COMMUNITY_LIST_XQUERY = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType')"
-		+ "  where $x//CONFIGURATION/context[./@type='community' or ./@type='ri']"
-		+ "  and  $x//CONFIGURATION/context/param[./@name='status']/text() != 'hidden'"
-		+ "  return $x//CONFIGURATION/context/@id/string()";
-
 	/**
 	 * associates to each result the set of community contexts they are associated to; associates to each target of a
 	 * relation with allowed semantics the set of community context it could possibly inherit from the source of the
@@ -64,7 +61,7 @@ public class PrepareResultCommunitySetStep1 {
 			.toString(
 				PrepareResultCommunitySetStep1.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromsemrel/input_preparecommunitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/input_preparecommunitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -88,10 +85,10 @@ public class PrepareResultCommunitySetStep1 {
 		final List<String> allowedsemrel = Arrays.asList(parser.get("allowedsemrels").split(";"));
 		log.info("allowedSemRel: {}", new Gson().toJson(allowedsemrel));
 
-		final String isLookupUrl = parser.get("isLookUpUrl");
+		final String baseURL = parser.get("baseURL");
-		log.info("isLookupUrl: {}", isLookupUrl);
+		log.info("baseURL: {}", baseURL);
 
-		final List<String> communityIdList = getCommunityList(isLookupUrl);
+		final List<String> communityIdList = getCommunityList(baseURL);
 		log.info("communityIdList: {}", new Gson().toJson(communityIdList));
 
 		final String resultType = resultClassName.substring(resultClassName.lastIndexOf(".") + 1).toLowerCase();
@@ -159,9 +156,8 @@ public class PrepareResultCommunitySetStep1 {
 			.json(outputResultPath);
 	}
 
-	public static List<String> getCommunityList(final String isLookupUrl) throws ISLookUpException {
+	public static List<String> getCommunityList(final String baseURL) throws IOException {
-		ISLookUpService isLookUp = ISLookupClientFactory.getLookUpService(isLookupUrl);
+		return Utils.getCommunityIdList(baseURL);
-		return isLookUp.quickSearchProfile(COMMUNITY_LIST_XQUERY);
 	}
 
 }

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromsemrel/PrepareResultCommunitySetStep2.java

@@ -31,7 +31,7 @@ public class PrepareResultCommunitySetStep2 {
 			.toString(
 				PrepareResultCommunitySetStep2.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromsemrel/input_preparecommunitytoresult2_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/input_preparecommunitytoresult2_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromsemrel/SparkResultToCommunityThroughSemRelJob.java

@@ -33,7 +33,7 @@ public class SparkResultToCommunityThroughSemRelJob {
 			.toString(
 				SparkResultToCommunityThroughSemRelJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttocommunityfromsemrel/input_communitytoresult_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/input_communitytoresult_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -100,6 +100,12 @@ public class SparkResultToCommunityThroughSemRelJob {
 			.mode(SaveMode.Overwrite)
 			.option("compression", "gzip")
 			.json(outputPath);
+
+		readPath(spark, outputPath, resultClazz)
+			.write()
+			.mode(SaveMode.Overwrite)
+			.option("compression", "gzip")
+			.json(inputPath);
 	}
 
 	private static <R extends Result> MapFunction<Tuple2<R, ResultCommunityList>, R> contextUpdaterFn() {

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttoorganizationfrominstrepo/AppendNewRelations.java

@@ -0,0 +1,68 @@
+
+package eu.dnetlib.dhp.resulttoorganizationfrominstrepo;
+
+import static eu.dnetlib.dhp.PropagationConstant.*;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.schema.oaf.Relation;
+
+/**
+ * @author miriam.baglioni
+ * @Date 09/12/23
+ */
+public class AppendNewRelations implements Serializable {
+
+	private static final Logger log = LoggerFactory.getLogger(AppendNewRelations.class);
+
+	public static void main(String[] args) throws Exception {
+
+		String jsonConfiguration = IOUtils
+			.toString(
+				AppendNewRelations.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/input_newrelation_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = isSparkSessionManaged(parser);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		String inputPath = parser.get("sourcePath");
+		log.info("inputPath: {}", inputPath);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath: {}", outputPath);
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> appendNewRelation(spark, inputPath, outputPath));
+	}
+
+	private static void appendNewRelation(SparkSession spark, String inputPath, String outputPath) {
+
+		readPath(spark, inputPath + "publication/relation", Relation.class)
+			.union(readPath(spark, inputPath + "dataset/relation", Relation.class))
+			.union(readPath(spark, inputPath + "otherresearchproduct/relation", Relation.class))
+			.union(readPath(spark, inputPath + "software/relation", Relation.class))
+			.write()
+			.mode(SaveMode.Append)
+			.option("compression", "gzip")
+			.json(outputPath);
+	}
+
+}

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttoorganizationfrominstrepo/PrepareResultInstRepoAssociation.java

@@ -40,7 +40,7 @@ public class PrepareResultInstRepoAssociation {
 			.toString(
 				PrepareResultInstRepoAssociation.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttoorganizationfrominstrepo/input_prepareresultorg_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/input_prepareresultorg_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -52,10 +52,13 @@ public class PrepareResultInstRepoAssociation {
 		String inputPath = parser.get("sourcePath");
 		log.info("inputPath: {}", inputPath);
 
-		final String datasourceOrganizationPath = parser.get("datasourceOrganizationPath");
+		final String workingPath = parser.get("workingPath");
+		log.info("workingPath : {}", workingPath);
+
+		final String datasourceOrganizationPath = workingPath + "/preparedInfo/datasourceOrganization";
 		log.info("datasourceOrganizationPath {}: ", datasourceOrganizationPath);
 
-		final String alreadyLinkedPath = parser.get("alreadyLinkedPath");
+		final String alreadyLinkedPath = workingPath + "/preparedInfo/alreadyLinked";
 		log.info("alreadyLinkedPath {}: ", alreadyLinkedPath);
 
 		List<String> blacklist = Optional

dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttoorganizationfrominstrepo/SparkResultToOrganizationFromIstRepoJob.java

@@ -47,7 +47,7 @@ public class SparkResultToOrganizationFromIstRepoJob {
 			.toString(
 				SparkResultToOrganizationFromIstRepoJob.class
 					.getResourceAsStream(
-						"/eu/dnetlib/dhp/resulttoorganizationfrominstrepo/input_propagationresulaffiliationfrominstrepo_parameters.json"));
+						"/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/input_propagationresulaffiliationfrominstrepo_parameters.json"));
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 
@@ -119,7 +119,7 @@ public class SparkResultToOrganizationFromIstRepoJob {
 				"left_outer")
 			.flatMap(createRelationFn(), Encoders.bean(Relation.class))
 			.write()
-			.mode(SaveMode.Append)
+			.mode(SaveMode.Overwrite)
 			.option("compression", "gzip")
 			.json(outputPath);
 	}

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/bulktag/datasourcemaster_parameters.json

@@ -1,32 +0,0 @@
-[
-	{
-		"paramName": "p",
-		"paramLongName": "hdfsPath",
-		"paramDescription": "the path where storing the sequential file",
-		"paramRequired": true
-	},
-	{
-		"paramName": "nn",
-		"paramLongName": "hdfsNameNode",
-		"paramDescription": "the name node on hdfs",
-		"paramRequired": true
-	},
-	{
-		"paramName": "pgurl",
-		"paramLongName": "postgresUrl",
-		"paramDescription": "postgres url, example: jdbc:postgresql://localhost:5432/testdb",
-		"paramRequired": true
-	},
-	{
-		"paramName": "pguser",
-		"paramLongName": "postgresUser",
-		"paramDescription": "postgres user",
-		"paramRequired": false
-	},
-	{
-		"paramName": "pgpasswd",
-		"paramLongName": "postgresPassword",
-		"paramDescription": "postgres password",
-		"paramRequired": false
-	}
-]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/bulktag/input_eoscTag_parameters.json

@@ -1,21 +0,0 @@
-[
-  {
-    "paramName":"s",
-    "paramLongName":"sourcePath",
-    "paramDescription": "the path of the sequencial file to read",
-    "paramRequired": true
-  },
-  {
-    "paramName": "wp",
-    "paramLongName": "workingPath",
-    "paramDescription": "the path used to store temporary output files",
-    "paramRequired": true
-  },
-  {
-    "paramName": "ssm",
-    "paramLongName": "isSparkSessionManaged",
-    "paramDescription": "true if the spark session is managed, false otherwise",
-    "paramRequired": false
-  }
-
-]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/bulktag/input_eosc_bulkTag_parameters.json

@@ -1,41 +0,0 @@
-[
-
-  {
-    "paramName":"s",
-    "paramLongName":"sourcePath",
-    "paramDescription": "the path of the sequencial file to read",
-    "paramRequired": true
-  },
-  {
-    "paramName": "dmp",
-    "paramLongName":"datasourceMapPath",
-    "paramDescription": "the path where the association datasource master has been stored",
-    "paramRequired": true
-  },
-  {
-    "paramName":"tn",
-    "paramLongName":"resultTableName",
-    "paramDescription": "the name of the result table we are currently working on",
-    "paramRequired": true
-  },
-  {
-    "paramName": "wp",
-    "paramLongName": "workingPath",
-    "paramDescription": "the path used to store temporary output files",
-    "paramRequired": true
-  },
-  {
-    "paramName": "ssm",
-    "paramLongName": "isSparkSessionManaged",
-    "paramDescription": "true if the spark session is managed, false otherwise",
-    "paramRequired": false
-  },
-  {
-
-    "paramName": "rt",
-    "paramLongName": "resultType",
-    "paramDescription": "the result type",
-    "paramRequired": true
-  }
-
-]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/entitytoorganizationfromsemrel/oozie_app/workflow.xml

@@ -1,197 +0,0 @@
-<workflow-app name="affiliation_from_semrel_propagation" xmlns="uri:oozie:workflow:0.5">
-    <parameters>
-        <property>
-            <name>sourcePath</name>
-            <description>the source path</description>
-        </property>
-        <property>
-            <name>outputPath</name>
-            <description>sets the outputPath</description>
-        </property>
-    </parameters>
-
-    <global>
-        <job-tracker>${jobTracker}</job-tracker>
-        <name-node>${nameNode}</name-node>
-        <configuration>
-            <property>
-                <name>oozie.action.sharelib.for.spark</name>
-                <value>${oozieActionShareLibForSpark2}</value>
-            </property>
-        </configuration>
-    </global>
-
-    <start to="resume_from"/>
-
-    <kill name="Kill">
-        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
-    </kill>
-
-    <decision name="resume_from">
-        <switch>
-            <case to="prepare_info">${wf:conf('resumeFrom') eq 'PrepareInfo'}</case>
-            <default to="reset_outputpath"/> <!-- first action to be done when downloadDump is to be performed -->
-        </switch>
-    </decision>
-
-    <action name="reset_outputpath">
-        <fs>
-            <delete path="${outputPath}"/>
-            <mkdir path="${outputPath}"/>
-        </fs>
-        <ok to="copy_entities"/>
-        <error to="Kill"/>
-    </action>
-
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_publication"/>
-        <path start="copy_dataset"/>
-        <path start="copy_orp"/>
-        <path start="copy_software"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_publication">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/publication</arg>
-            <arg>${nameNode}/${outputPath}/publication</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_dataset">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/dataset</arg>
-            <arg>${nameNode}/${outputPath}/dataset</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_orp">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/otherresearchproduct</arg>
-            <arg>${nameNode}/${outputPath}/otherresearchproduct</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_software">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/software</arg>
-            <arg>${nameNode}/${outputPath}/software</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="wait" to="prepare_info"/>
-
-
-    <action name="prepare_info">
-        <spark xmlns="uri:oozie:spark-action:0.2">
-            <master>yarn</master>
-            <mode>cluster</mode>
-            <name>PrepareResultOrganizationAssociation</name>
-            <class>eu.dnetlib.dhp.entitytoorganizationfromsemrel.PrepareInfo</class>
-            <jar>dhp-enrichment-${projectVersion}.jar</jar>
-            <spark-opts>
-                --executor-cores=${sparkExecutorCores}
-                --executor-memory=${sparkExecutorMemory}
-                --driver-memory=${sparkDriverMemory}
-                --conf spark.extraListeners=${spark2ExtraListeners}
-                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
-                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
-                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
-            </spark-opts>
-            <arg>--graphPath</arg><arg>${sourcePath}</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
-            <arg>--leavesPath</arg><arg>${workingDir}/preparedInfo/leavesPath</arg>
-            <arg>--childParentPath</arg><arg>${workingDir}/preparedInfo/childParentPath</arg>
-            <arg>--resultOrgPath</arg><arg>${workingDir}/preparedInfo/resultOrgPath</arg>
-            <arg>--projectOrganizationPath</arg><arg>${workingDir}/preparedInfo/projectOrganizationPath</arg>
-            <arg>--relationPath</arg><arg>${workingDir}/preparedInfo/relation</arg>
-        </spark>
-        <ok to="apply_resulttoorganization_propagation"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="apply_resulttoorganization_propagation">
-        <spark xmlns="uri:oozie:spark-action:0.2">
-            <master>yarn</master>
-            <mode>cluster</mode>
-            <name>resultToOrganizationFromSemRel</name>
-            <class>eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkResultToOrganizationFromSemRel</class>
-            <jar>dhp-enrichment-${projectVersion}.jar</jar>
-            <spark-opts>
-                --executor-cores=${sparkExecutorCores}
-                --executor-memory=${sparkExecutorMemory}
-                --driver-memory=${sparkDriverMemory}
-                --conf spark.extraListeners=${spark2ExtraListeners}
-                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
-                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
-                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
-                --conf spark.dynamicAllocation.enabled=true
-                --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
-                --conf spark.sql.shuffle.partitions=3840
-            </spark-opts>
-            <arg>--relationPath</arg><arg>${workingDir}/preparedInfo/relation</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
-            <arg>--leavesPath</arg><arg>${workingDir}/preparedInfo/leavesPath</arg>
-            <arg>--childParentPath</arg><arg>${workingDir}/preparedInfo/childParentPath</arg>
-            <arg>--resultOrgPath</arg><arg>${workingDir}/preparedInfo/resultOrgPath</arg>
-            <arg>--projectOrganizationPath</arg><arg>${workingDir}/preparedInfo/projectOrganizationPath</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
-            <arg>--workingDir</arg><arg>${workingDir}/working</arg>
-            <arg>--iterations</arg><arg>${iterations}</arg>
-        </spark>
-        <ok to="End"/>
-        <error to="Kill"/>
-    </action>
-
-
-
-    <end name="End"/>
-
-</workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/main/job.properties

@@ -0,0 +1,28 @@
+sourcePath=/tmp/beta_provision/graph/09_graph_dedup_enriched
+resumeFrom=CountryPropagation
+allowedsemrelsorcidprop=isSupplementedBy;isSupplementTo
+allowedsemrelsresultproject=isSupplementedBy;isSupplementTo
+allowedsemrelscommunitysemrel=isSupplementedBy;isSupplementTo
+datasourceWhitelistForCountryPropagation=10|opendoar____::16e6a3326dd7d868cbc926602a61e4d0;10|openaire____::fdb035c8b3e0540a8d9a561a6c44f4de;10|eurocrisdris::fe4903425d9040f680d8610d9079ea14;10|openaire____::5b76240cc27a58c6f7ceef7d8c36660e;10|openaire____::172bbccecf8fca44ab6a6653e84cb92a;10|openaire____::149c6590f8a06b46314eed77bfca693f;10|eurocrisdris::a6026877c1a174d60f81fd71f62df1c1;10|openaire____::4692342f0992d91f9e705c26959f09e0;10|openaire____::8d529dbb05ec0284662b391789e8ae2a;10|openaire____::345c9d171ef3c5d706d08041d506428c;10|opendoar____::1c1d4df596d01da60385f0bb17a4a9e0;10|opendoar____::7a614fd06c325499f1680b9896beedeb;10|opendoar____::1ee3dfcd8a0645a25a35977997223d22;10|opendoar____::d296c101daa88a51f6ca8cfc1ac79b50;10|opendoar____::798ed7d4ee7138d49b8828958048130a;10|openaire____::c9d2209ecc4d45ba7b4ca7597acb88a2;10|eurocrisdris::c49e0fe4b9ba7b7fab717d1f0f0a674d;10|eurocrisdris::9ae43d14471c4b33661fedda6f06b539;10|eurocrisdris::432ca599953ff50cd4eeffe22faf3e48
+#allowedtypes=pubsrepository::institutional
+allowedtypes=Institutional
+outputPath=/tmp/miriam/enrichment_one_step
+pathMap ={"author":"$['author'][*]['fullname']", \
+  "title":"$['title'][*]['value']",\
+  "orcid":"$['author'][*]['pid'][*][?(@['qualifier']['classid']=='orcid')]['value']" ,\
+  "orcid_pending":"$['author'][*]['pid'][*][?(@['qualifier']['classid']=='orcid_pending')]['value']" ,\
+  "contributor" : "$['contributor'][*]['value']",\
+  "description" : "$['description'][*]['value']",\
+  "subject" :"$['subject'][*]['value']" , \
+  "fos" : "$['subject'][?(@['qualifier']['classid']=='FOS')].value" ,\
+  "sdg" : "$['subject'][?(@['qualifier']['classid']=='SDG')].value",\
+  "journal":"$['journal'].name",\
+  "hostedby":"$['instance'][*]['hostedby']['key']",\
+  "collectedfrom":"$['instance'][*]['collectedfrom']['key']",\
+  "publisher":"$['publisher'].value",\
+  "publicationyear":"$['dateofacceptance'].value"}
+blacklist=empty
+allowedpids=orcid;orcid_pending
+baseURL = https://services.openaire.eu/openaire/community/
+iterations=1
+

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/main/oozie_app/config-default.xml

@@ -0,0 +1,30 @@
+<configuration>
+    <property>
+        <name>jobTracker</name>
+        <value>yarnRM</value>
+    </property>
+    <property>
+        <name>nameNode</name>
+        <value>hdfs://nameservice1</value>
+    </property>
+    <property>
+        <name>oozie.use.system.libpath</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>hiveMetastoreUris</name>
+        <value>thrift://iis-cdh5-test-m3.ocean.icm.edu.pl:9083</value>
+    </property>
+    <property>
+        <name>hiveJdbcUrl</name>
+        <value>jdbc:hive2://iis-cdh5-test-m3.ocean.icm.edu.pl:10000</value>
+    </property>
+    <property>
+        <name>hiveDbName</name>
+        <value>openaire</value>
+    </property>
+    <property>
+        <name>oozie.launcher.mapreduce.user.classpath.first</name>
+        <value>true</value>
+    </property>
+</configuration>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/main/oozie_app/import.txt

@@ -0,0 +1,10 @@
+## This is a classpath-based import file (this header is required)
+orcid_propagation classpath eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/oozie_app
+bulk_tagging classpath  eu/dnetlib/dhp/wf/subworkflows/bulktag/oozie_app
+affiliation_inst_repo classpath  eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/oozie_app
+entity_semantic_relation classpath eu/dnetlib/dhp/wf/subworkflows/entitytoorganizationfromsemrel/oozie_app
+community_organization classpath eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromorganization/oozie_app
+result_project classpath eu/dnetlib/dhp/wf/subworkflows/projecttoresult/oozie_app
+community_project classpath eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromproject/oozie_app
+community_sem_rel classpath eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/oozie_app
+country_propagation classpath eu/dnetlib/dhp/wf/subworkflows/countrypropagation/oozie_app

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/main/oozie_app/workflow.xml

@@ -0,0 +1,324 @@
+<workflow-app name="enrichment_main" xmlns="uri:oozie:workflow:0.5">
+
+    <parameters>
+        <property>
+            <name>sourcePath</name>
+            <description>the source path</description>
+        </property>
+        <property>
+            <name>allowedsemrelsorcidprop</name>
+            <description>the semantic relationships allowed for propagation</description>
+        </property>
+        <property>
+            <name>allowedsemrelsresultproject</name>
+            <description>the allowed semantics </description>
+        </property>
+        <property>
+            <name>allowedsemrelscommunitysemrel</name>
+            <description>the semantic relationships allowed for propagation</description>
+        </property>
+        <property>
+            <name>datasourceWhitelistForCountryPropagation</name>
+            <description>the white list</description>
+        </property>
+        <property>
+            <name>allowedtypes</name>
+            <description>the allowed types</description>
+        </property>
+        <property>
+            <name>outputPath</name>
+            <description>the output path</description>
+        </property>
+        <property>
+            <name>pathMap</name>
+            <description>the json path associated to each selection field</description>
+        </property>
+        <property>
+            <name>blacklist</name>
+            <description>list of datasources in blacklist for the affiliation from instrepo propagation</description>
+        </property>
+
+        <property>
+            <name>hiveDbName</name>
+            <description>the target hive database name</description>
+        </property>
+        <property>
+            <name>hiveJdbcUrl</name>
+            <description>hive server jdbc url</description>
+        </property>
+        <property>
+            <name>hiveMetastoreUris</name>
+            <description>hive server metastore URIs</description>
+        </property>
+        <property>
+            <name>sparkDriverMemory</name>
+            <description>memory for driver process</description>
+        </property>
+        <property>
+            <name>sparkExecutorMemory</name>
+            <description>memory for individual executor</description>
+        </property>
+        <property>
+            <name>sparkExecutorCores</name>
+            <description>number of cores used by single executor</description>
+        </property>
+        <property>
+            <name>oozieActionShareLibForSpark2</name>
+            <description>oozie action sharelib for spark 2.*</description>
+        </property>
+        <property>
+            <name>spark2ExtraListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
+            <description>spark 2.* extra listeners classname</description>
+        </property>
+        <property>
+            <name>spark2SqlQueryExecutionListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
+            <description>spark 2.* sql query execution listeners classname</description>
+        </property>
+        <property>
+            <name>spark2YarnHistoryServerAddress</name>
+            <description>spark 2.* yarn history server address</description>
+        </property>
+        <property>
+            <name>spark2EventLogDir</name>
+            <description>spark 2.* event log dir location</description>
+        </property>
+    </parameters>
+
+    <global>
+        <job-tracker>${jobTracker}</job-tracker>
+        <name-node>${nameNode}</name-node>
+        <configuration>
+            <property>
+                <name>mapreduce.job.queuename</name>
+                <value>${queueName}</value>
+            </property>
+            <property>
+                <name>oozie.launcher.mapred.job.queue.name</name>
+                <value>${oozieLauncherQueueName}</value>
+            </property>
+            <property>
+                <name>oozie.action.sharelib.for.spark</name>
+                <value>${oozieActionShareLibForSpark2}</value>
+            </property>
+
+        </configuration>
+    </global>
+
+    <start to="resumeFrom"/>
+
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+
+    <decision name="resumeFrom">
+        <switch>
+            <case to="bulk_tagging">${wf:conf('resumeFrom') eq 'BulkTagging'}</case>
+            <case to="affiliation_inst_repo">${wf:conf('resumeFrom') eq 'AffiliationInstitutionalRepository'}</case>
+            <case to="entity_semantic_relation">${wf:conf('resumeFrom') eq 'AffiliationSemanticRelation'}</case>
+            <case to="community_organization">${wf:conf('resumeFrom') eq 'CommunityOrganization'}</case>
+            <case to="result_project">${wf:conf('resumeFrom') eq 'ResultProject'}</case>
+            <case to="community_project">${wf:conf('resumeFrom') eq 'CommunityProject'}</case>
+            <case to="community_sem_rel">${wf:conf('resumeFrom') eq 'CommunitySemanticRelation'}</case>
+            <case to="country_propagation">${wf:conf('resumeFrom') eq 'CountryPropagation'}</case>
+            <default to="orcid_propagation"/>
+        </switch>
+    </decision>
+
+
+    <action name="orcid_propagation">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/orcid_propagation
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${sourcePath}</value>
+                </property>
+                <property>
+                    <name>allowedsemrels</name>
+                    <value>${allowedsemrelsorcidprop}</value>
+                </property>
+                <property>
+                    <name>outputPath</name>
+                    <value>${outputPath}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="bulk_tagging" />
+        <error to="Kill" />
+    </action>
+
+    <action name="bulk_tagging">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/bulk_tagging
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>baseURL</name>
+                    <value>${baseURL}</value>
+                </property>
+                <property>
+                    <name>pathMap</name>
+                    <value>${pathMap}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="affiliation_inst_repo" />
+        <error to="Kill" />
+    </action>
+
+    <action name="affiliation_inst_repo">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/affiliation_inst_repo
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>blacklist</name>
+                    <value>${blacklist}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="entity_semantic_relation" />
+        <error to="Kill" />
+    </action>
+
+    <action name="entity_semantic_relation">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/entity_semantic_relation
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>iterations</name>
+                    <value>${iterations}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="community_organization" />
+        <error to="Kill" />
+    </action>
+
+    <action name="community_organization">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/community_organization
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>baseURL</name>
+                    <value>${baseURL}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="result_project" />
+        <error to="Kill" />
+    </action>
+
+    <action name="result_project">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/result_project
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>allowedsemrels</name>
+                    <value>${allowedsemrelsresultproject}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="community_project" />
+        <error to="Kill" />
+    </action>
+
+    <action name="community_project">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/community_project
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="community_sem_rel" />
+        <error to="Kill" />
+    </action>
+
+    <action name="community_sem_rel">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/community_sem_rel
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>allowedsemrels</name>
+                    <value>${allowedsemrelscommunitysemrel}</value>
+                </property>
+                <property>
+                    <name>baseURL</name>
+                    <value>${baseURL}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="country_propagation" />
+        <error to="Kill" />
+    </action>
+
+    <action name="country_propagation">
+        <sub-workflow>
+            <app-path>${wf:appPath()}/country_propagation
+            </app-path>
+            <propagate-configuration/>
+            <configuration>
+                <property>
+                    <name>sourcePath</name>
+                    <value>${outputPath}</value>
+                </property>
+                <property>
+                    <name>whitelist</name>
+                    <value>${datasourceWhitelistForCountryPropagation}</value>
+                </property>
+                <property>
+                    <name>allowedtypes</name>
+                    <value>${allowedtypes}</value>
+                </property>
+            </configuration>
+        </sub-workflow>
+        <ok to="End" />
+        <error to="Kill" />
+    </action>
+
+    <end name="End"/>
+
+</workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/bulktag/oozie_app/workflow.xml

@@ -8,14 +8,11 @@
             <name>pathMap</name>
             <description>the json path associated to each selection field</description>
         </property>
-        <property>
-            <name>outputPath</name>
-            <description>the output path</description>
-        </property>
         <property>
             <name>baseURL</name>
-            <description>the community API base URL</description>
+            <description>The URL to access the community APIs</description>
         </property>
+
     </parameters>
 
     <global>
@@ -37,63 +34,18 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="exec_bulktag"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="copy_wait" to="exec_bulktag"/>
-
     <action name="exec_bulktag">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn-cluster</master>
             <mode>cluster</mode>
-            <name>bulkTagging-result</name>
+            <name>bulkTagging-publication</name>
             <class>eu.dnetlib.dhp.bulktag.SparkBulkTagJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
@@ -107,7 +59,7 @@
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/bulktag/</arg>
             <arg>--pathMap</arg><arg>${pathMap}</arg>
             <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
@@ -115,6 +67,8 @@
         <error to="Kill"/>
     </action>
 
+
+
     <end name="End"/>
 
 </workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/countrypropagation/input_countrypropagation_parameters.json

@@ -29,4 +29,4 @@
     "paramDescription": "true if the spark session is managed, false otherwise",
     "paramRequired": false
   }
-]
+]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/countrypropagation/oozie_app/workflow.xml

@@ -12,11 +12,6 @@
             <name>allowedtypes</name>
             <description>the allowed types</description>
         </property>
-        <property>
-            <name>outputPath</name>
-            <description>the output path</description>
-        </property>
-
     </parameters>
 
     <global>
@@ -38,57 +33,13 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="prepare_datasource_country_association"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="copy_wait" to="prepare_datasource_country_association"/>
 
     <action name="prepare_datasource_country_association">
         <spark xmlns="uri:oozie:spark-action:0.2">
@@ -112,18 +63,18 @@
             <arg>--allowedtypes</arg><arg>${allowedtypes}</arg>
             <arg>--outputPath</arg><arg>${workingDir}/preparedInfo</arg>
         </spark>
-        <ok to="fork_join_prepare_result_country"/>
+        <ok to="fork_prepare_result_country"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="fork_join_prepare_result_country">
+    <fork name="fork_prepare_result_country">
-        <path start="join_prepareresult_publication"/>
+        <path start="prepareresult_publication"/>
-        <path start="join_prepareresult_dataset"/>
+        <path start="prepareresult_dataset"/>
-        <path start="join_prepareresult_otherresearchproduct"/>
+        <path start="prepareresult_otherresearchproduct"/>
-        <path start="join_prepareresult_software"/>
+        <path start="prepareresult_software"/>
     </fork>
 
-    <action name="join_prepareresult_publication">
+    <action name="prepareresult_publication">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -153,7 +104,7 @@
         <error to="Kill"/>
     </action>
 
-    <action name="join_prepareresult_dataset">
+    <action name="prepareresult_dataset">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -183,7 +134,7 @@
         <error to="Kill"/>
     </action>
 
-    <action name="join_prepareresult_otherresearchproduct">
+    <action name="prepareresult_otherresearchproduct">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -213,7 +164,7 @@
         <error to="Kill"/>
     </action>
 
-    <action name="join_prepareresult_software">
+    <action name="prepareresult_software">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -243,16 +194,16 @@
         <error to="Kill"/>
     </action>
 
-    <join name="wait_prepare" to="fork_join_apply_country_propagation"/>
+    <join name="wait_prepare" to="fork_apply_country_propagation"/>
 
-    <fork name="fork_join_apply_country_propagation">
+    <fork name="fork_apply_country_propagation">
-        <path start="join_propagation_publication"/>
+        <path start="propagation_publication"/>
-        <path start="join_propagation_dataset"/>
+        <path start="propagation_dataset"/>
-        <path start="join_propagation_otherresearchproduct"/>
+        <path start="propagation_otherresearchproduct"/>
-        <path start="join_propagation_software"/>
+        <path start="propagation_software"/>
     </fork>
 
-    <action name="join_propagation_publication">
+    <action name="propagation_publication">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -275,13 +226,13 @@
             <arg>--sourcePath</arg><arg>${sourcePath}/publication</arg>
             <arg>--preparedInfoPath</arg><arg>${workingDir}/publication</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/publication</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/country/publication</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
-    <action name="join_propagation_dataset">
+    <action name="propagation_dataset">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -304,13 +255,13 @@
             <arg>--sourcePath</arg><arg>${sourcePath}/dataset</arg>
             <arg>--preparedInfoPath</arg><arg>${workingDir}/dataset</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/dataset</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/country/dataset</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
-    <action name="join_propagation_otherresearchproduct">
+    <action name="propagation_otherresearchproduct">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -333,13 +284,13 @@
             <arg>--sourcePath</arg><arg>${sourcePath}/otherresearchproduct</arg>
             <arg>--preparedInfoPath</arg><arg>${workingDir}/otherresearchproduct</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/otherresearchproduct</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/country/otherresearchproduct</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
-    <action name="join_propagation_software">
+    <action name="propagation_software">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
@@ -362,14 +313,21 @@
             <arg>--sourcePath</arg><arg>${sourcePath}/software</arg>
             <arg>--preparedInfoPath</arg><arg>${workingDir}/software</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/software</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/country/software</arg>
-        </spark>
+                    </spark>
         <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
-    <join name="wait" to="End"/>
+    <join name="wait" to="reset_workingDir"/>
-
+    <action name="reset_workingDir">
+        <fs>
+            <delete path="${workingDir}"/>
+            <mkdir path="${workingDir}"/>
+        </fs>
+        <ok to="End"/>
+        <error to="Kill"/>
+    </action>
     <end name="End"/>
 
 </workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/entitytoorganizationfromsemrel/oozie_app/workflow.xml

@@ -0,0 +1,101 @@
+<workflow-app name="affiliation_from_semrel_propagation" xmlns="uri:oozie:workflow:0.5">
+    <parameters>
+        <property>
+            <name>sourcePath</name>
+            <description>the source path</description>
+        </property>
+
+    </parameters>
+
+    <global>
+        <job-tracker>${jobTracker}</job-tracker>
+        <name-node>${nameNode}</name-node>
+        <configuration>
+            <property>
+                <name>oozie.action.sharelib.for.spark</name>
+                <value>${oozieActionShareLibForSpark2}</value>
+            </property>
+        </configuration>
+    </global>
+
+    <start to="reset_outputpath"/>
+
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+
+    <action name="reset_outputpath">
+        <fs>
+            <delete path="${workingDir}"/>
+            <mkdir path="${workingDir}"/>
+        </fs>
+        <ok to="prepare_info"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="prepare_info">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>PrepareResultProjectOrganizationAssociation</name>
+            <class>eu.dnetlib.dhp.entitytoorganizationfromsemrel.PrepareInfo</class>
+            <jar>dhp-enrichment-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+            </spark-opts>
+            <arg>--graphPath</arg><arg>${sourcePath}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--leavesPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/leavesPath</arg>
+            <arg>--childParentPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/childParentPath</arg>
+            <arg>--resultOrgPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/resultOrgPath</arg>
+            <arg>--projectOrganizationPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/projectOrganizationPath</arg>
+            <arg>--relationPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/relation</arg>
+        </spark>
+        <ok to="apply_resulttoorganization_propagation"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="apply_resulttoorganization_propagation">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>entityToOrganizationFromSemRel</name>
+            <class>eu.dnetlib.dhp.entitytoorganizationfromsemrel.SparkEntityToOrganizationFromSemRel</class>
+            <jar>dhp-enrichment-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.dynamicAllocation.enabled=true
+                --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
+                --conf spark.sql.shuffle.partitions=3840
+            </spark-opts>
+            <arg>--relationPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/relation</arg>
+            <arg>--outputPath</arg><arg>${sourcePath}/relation</arg>
+            <arg>--leavesPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/leavesPath</arg>
+            <arg>--childParentPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/childParentPath</arg>
+            <arg>--resultOrgPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/resultOrgPath</arg>
+            <arg>--projectOrganizationPath</arg><arg>${workingDir}/entitiesSemanticRelation/preparedInfo/projectOrganizationPath</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/entitiesSemanticRelation/working</arg>
+            <arg>--iterations</arg><arg>${iterations}</arg>
+        </spark>
+        <ok to="End"/>
+        <error to="Kill"/>
+    </action>
+
+
+
+    <end name="End"/>
+
+</workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/input_orcidtoresult_parameters.json

@@ -11,12 +11,6 @@
     "paramDescription": "true if the new version of the graph must be saved",
     "paramRequired": false
   },
-  {
-    "paramName":"h",
-    "paramLongName":"hive_metastore_uris",
-    "paramDescription": "the hive metastore uris",
-    "paramRequired": true
-  },
   {
     "paramName": "out",
     "paramLongName": "outputPath",

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/orcidtoresultfromsemrel/oozie_app/workflow.xml

@@ -225,8 +225,8 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--sourcePath</arg><arg>${workingDir}/preparedInfo/targetOrcidAssoc</arg>
+            <arg>--sourcePath</arg><arg>${workingDir}/orcidprop</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
         </spark>
         <ok to="fork-join-exec-propagation"/>
         <error to="Kill"/>
@@ -261,9 +261,8 @@
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
                 --conf spark.sql.shuffle.partitions=3840
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/publication</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
             <arg>--outputPath</arg><arg>${outputPath}/publication</arg>
         </spark>
@@ -292,9 +291,8 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/dataset</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
             <arg>--outputPath</arg><arg>${outputPath}/dataset</arg>
         </spark>
@@ -323,9 +321,8 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/otherresearchproduct</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
             <arg>--outputPath</arg><arg>${outputPath}/otherresearchproduct</arg>
         </spark>
@@ -354,9 +351,8 @@
                 --conf spark.hadoop.mapreduce.map.speculative=false
                 --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
-            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/preparedInfo/mergedOrcidAssoc</arg>
+            <arg>--possibleUpdatesPath</arg><arg>${workingDir}/orcidprop/mergedOrcidAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/software</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
             <arg>--outputPath</arg><arg>${outputPath}/software</arg>
         </spark>
@@ -365,6 +361,15 @@
     </action>
     
     <join name="wait2" to="End"/>
+
+<!--    <action name="reset_workingDir">-->
+<!--        <fs>-->
+<!--            <delete path="${workingDir}"/>-->
+<!--            <mkdir path="${workingDir}"/>-->
+<!--        </fs>-->
+<!--        <ok to="End"/>-->
+<!--        <error to="Kill"/>-->
+<!--    </action>-->
     
     <end name="End"/>
     

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/projecttoresult/oozie_app/workflow.xml

@@ -8,10 +8,7 @@
             <name>allowedsemrels</name>
             <description>the allowed semantics </description>
         </property>
-       <property>
+
-           <name>outputPath</name>
-           <description>the output path</description>
-       </property>
     </parameters>
 
     <global>
@@ -33,98 +30,13 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="prepare_project_results_association"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_publication"/>
-        <path start="copy_dataset"/>
-        <path start="copy_orp"/>
-        <path start="copy_software"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_publication">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/publication</arg>
-            <arg>${nameNode}/${outputPath}/publication</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_dataset">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/dataset</arg>
-            <arg>${nameNode}/${outputPath}/dataset</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_orp">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/otherresearchproduct</arg>
-            <arg>${nameNode}/${outputPath}/otherresearchproduct</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_software">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/software</arg>
-            <arg>${nameNode}/${outputPath}/software</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="wait" to="prepare_project_results_association"/>
-    
     <action name="prepare_project_results_association">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
@@ -144,8 +56,8 @@
             <arg>--sourcePath</arg><arg>${sourcePath}/relation</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
-            <arg>--potentialUpdatePath</arg><arg>${workingDir}/preparedInfo/potentialUpdates</arg>
+            <arg>--potentialUpdatePath</arg><arg>${workingDir}/resultproject/preparedInfo/potentialUpdates</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/resultproject/preparedInfo/alreadyLinked</arg>
         </spark>
         <ok to="apply_propagation"/>
         <error to="Kill"/>
@@ -169,16 +81,22 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${sourcePath}/relation</arg>
-            <arg>--potentialUpdatePath</arg><arg>${workingDir}/preparedInfo/potentialUpdates</arg>
+            <arg>--potentialUpdatePath</arg><arg>${workingDir}/resultproject/preparedInfo/potentialUpdates</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/resultproject/preparedInfo/alreadyLinked</arg>
         </spark>
+        <ok to="reset_workingDir"/>
+        <error to="Kill"/>
+    </action>
+    <action name="reset_workingDir">
+        <fs>
+            <delete path="${workingDir}"/>
+            <mkdir path="${workingDir}"/>
+        </fs>
         <ok to="End"/>
         <error to="Kill"/>
     </action>
-
     <end name="End"/>
 
 </workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromorganization/oozie_app/workflow.xml

@@ -4,13 +4,9 @@
             <name>sourcePath</name>
             <description>the source path</description>
         </property>
-        <property>
-            <name>outputPath</name>
-            <description>the output path</description>
-        </property>
         <property>
             <name>baseURL</name>
-            <description>the community API base URL</description>
+            <description>the baseURL from where to reach the community APIs</description>
         </property>
     </parameters>
 
@@ -33,58 +29,13 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="prepare_result_communitylist"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="copy_wait" to="prepare_result_communitylist"/>
-
     <action name="prepare_result_communitylist">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
@@ -102,10 +53,11 @@
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
                 --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/relation</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/resultCommunityList</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communityorganization/preparedInfo/resultCommunityList</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
@@ -117,7 +69,7 @@
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
             <mode>cluster</mode>
-            <name>community2resultfromorganization</name>
+            <name>community2resultfromorganization-Publication</name>
             <class>eu.dnetlib.dhp.resulttocommunityfromorganization.SparkResultToCommunityFromOrganizationJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
@@ -132,16 +84,14 @@
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/resultCommunityList</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communityorganization/preparedInfo/resultCommunityList</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communityorganization/resulttocommunityfromorganization/</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>
     </action>
 
-
-
     <end name="End"/>
 
 </workflow-app>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromproject/oozie_app/workflow.xml

@@ -4,13 +4,9 @@
             <name>sourcePath</name>
             <description>the source path</description>
         </property>
-        <property>
-            <name>outputPath</name>
-            <description>the output path</description>
-        </property>
         <property>
             <name>baseURL</name>
-            <description>the community API base URL</description>
+            <description>the base URL to use to select the right community APIs</description>
         </property>
     </parameters>
 
@@ -30,61 +26,15 @@
     <kill name="Kill">
         <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
     </kill>
-
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="prepare_result_communitylist"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="copy_wait" to="prepare_result_communitylist"/>
-
     <action name="prepare_result_communitylist">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
@@ -106,7 +56,7 @@
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/relation</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/resultCommunityList</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitythroughproject/preparedInfo/resultCommunityList</arg>
             <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
         <ok to="exec-propagation"/>
@@ -132,9 +82,9 @@
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/resultCommunityList</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communitythroughproject/preparedInfo/resultCommunityList</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitythroughproject/</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/input_preparecommunitytoresult_parameters.json

@@ -1,7 +1,7 @@
 [
   {
-    "paramName":"is",
+    "paramName":"bu",
-    "paramLongName":"isLookUpUrl",
+    "paramLongName":"baseURL",
     "paramDescription": "URL of the isLookUp Service",
     "paramRequired": true
   },
@@ -36,9 +36,9 @@
     "paramRequired": true
   },
   {
-  "paramName":"tn",
+    "paramName":"tn",
-  "paramLongName":"resultTableName",
+    "paramLongName":"resultTableName",
-  "paramDescription": "the name of the result table we are currently working on",
+    "paramDescription": "the name of the result table we are currently working on",
-  "paramRequired": true
+    "paramRequired": true
- }
+  }
 ]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttocommunityfromsemrel/oozie_app/workflow.xml

@@ -9,7 +9,7 @@
             <description>the semantic relationships allowed for propagation</description>
         </property>
         <property>
-            <name>isLookUpUrl</name>
+            <name>baseURL</name>
             <description>the isLookup service endpoint</description>
         </property>
         <property>
@@ -26,66 +26,13 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="fork_prepare_assoc_step1"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <job-tracker>${jobTracker}</job-tracker>
-            <name-node>${nameNode}</name-node>
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <job-tracker>${jobTracker}</job-tracker>
-            <name-node>${nameNode}</name-node>
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <job-tracker>${jobTracker}</job-tracker>
-            <name-node>${nameNode}</name-node>
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <job-tracker>${jobTracker}</job-tracker>
-            <name-node>${nameNode}</name-node>
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="copy_wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="copy_wait" to="fork_prepare_assoc_step1"/>
-
     <fork name="fork_prepare_assoc_step1">
         <path start="join_prepare_publication"/>
         <path start="join_prepare_dataset"/>
@@ -101,8 +48,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.PrepareResultCommunitySetStep1</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -114,9 +63,9 @@
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/targetCommunityAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/targetCommunityAssoc</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
-            <arg>--isLookUpUrl</arg><arg>${isLookUpUrl}</arg>
+            <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
@@ -130,8 +79,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.PrepareResultCommunitySetStep1</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -143,9 +94,9 @@
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/targetCommunityAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/targetCommunityAssoc</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
-            <arg>--isLookUpUrl</arg><arg>${isLookUpUrl}</arg>
+            <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
@@ -159,8 +110,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.PrepareResultCommunitySetStep1</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -172,9 +125,9 @@
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/targetCommunityAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/targetCommunityAssoc</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
-            <arg>--isLookUpUrl</arg><arg>${isLookUpUrl}</arg>
+            <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
@@ -188,8 +141,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.PrepareResultCommunitySetStep1</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -201,9 +156,9 @@
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/targetCommunityAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/targetCommunityAssoc</arg>
             <arg>--allowedsemrels</arg><arg>${allowedsemrels}</arg>
-            <arg>--isLookUpUrl</arg><arg>${isLookUpUrl}</arg>
+            <arg>--baseURL</arg><arg>${baseURL}</arg>
         </spark>
         <ok to="wait"/>
         <error to="Kill"/>
@@ -219,8 +174,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.PrepareResultCommunitySetStep2</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -229,8 +186,8 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--sourcePath</arg><arg>${workingDir}/preparedInfo/targetCommunityAssoc</arg>
+            <arg>--sourcePath</arg><arg>${workingDir}/communitysemrel/preparedInfo/targetCommunityAssoc</arg>
-            <arg>--outputPath</arg><arg>${workingDir}/preparedInfo/mergedCommunityAssoc</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/mergedCommunityAssoc</arg>
         </spark>
         <ok to="fork-join-exec-propagation"/>
         <error to="Kill"/>
@@ -251,8 +208,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.SparkResultToCommunityThroughSemRelJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -261,12 +220,12 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/mergedCommunityAssoc</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/mergedCommunityAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/publication</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/publication</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/publication</arg>
-            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+
         </spark>
         <ok to="wait2"/>
         <error to="Kill"/>
@@ -280,8 +239,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.SparkResultToCommunityThroughSemRelJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -290,12 +251,12 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/mergedCommunityAssoc</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/mergedCommunityAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/dataset</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/dataset</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/dataset</arg>
-            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+
         </spark>
         <ok to="wait2"/>
         <error to="Kill"/>
@@ -309,8 +270,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.SparkResultToCommunityThroughSemRelJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -319,12 +282,12 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/mergedCommunityAssoc</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/mergedCommunityAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/otherresearchproduct</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/otherresearchproduct</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/otherresearchproduct</arg>
-            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+
         </spark>
         <ok to="wait2"/>
         <error to="Kill"/>
@@ -338,8 +301,10 @@
             <class>eu.dnetlib.dhp.resulttocommunityfromsemrel.SparkResultToCommunityThroughSemRelJob</class>
             <jar>dhp-enrichment-${projectVersion}.jar</jar>
             <spark-opts>
-                --executor-cores=${sparkExecutorCores}
+                --executor-cores=6
-                --executor-memory=${sparkExecutorMemory}
+                --executor-memory=5G
+                --conf spark.executor.memoryOverhead=3g
+                --conf spark.sql.shuffle.partitions=3284
                 --driver-memory=${sparkDriverMemory}
                 --conf spark.extraListeners=${spark2ExtraListeners}
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
@@ -348,12 +313,12 @@
                 --conf spark.dynamicAllocation.enabled=true
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
-            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo/mergedCommunityAssoc</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/communitysemrel/preparedInfo/mergedCommunityAssoc</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}/software</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/software</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/communitysemrel/software</arg>
-            <arg>--saveGraph</arg><arg>${saveGraph}</arg>
+
         </spark>
         <ok to="wait2"/>
         <error to="Kill"/>

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/input_newrelation_parameters.json

@@ -0,0 +1,20 @@
+[
+  {
+    "paramName":"s",
+    "paramLongName":"sourcePath",
+    "paramDescription": "the path of the sequencial file to read",
+    "paramRequired": true
+  },
+
+  {
+    "paramName": "ssm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "the path where prepared info have been stored",
+    "paramRequired": false
+  },{
+  "paramName": "o",
+  "paramLongName": "outputPath",
+  "paramDescription": "institutional repositories that should not be considered for the propagation",
+  "paramRequired": false
+}
+]

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/input_prepareresultorg_parameters.json

@@ -11,16 +11,11 @@
     "paramDescription": "the hive metastore uris",
     "paramRequired": true
   },
+
   {
-    "paramName":"dop",
+    "paramName":"wp",
-    "paramLongName":"datasourceOrganizationPath",
+    "paramLongName":"workingPath",
-    "paramDescription": "path where to store/find association from datasource and organization",
+    "paramDescription": "the working path",
-    "paramRequired": true
-  },
-  {
-    "paramName":"alp",
-    "paramLongName":"alreadyLinkedPath",
-    "paramDescription": "path where to store/find already linked results and organizations",
     "paramRequired": true
   },
   {

dhp-workflows/dhp-enrichment/src/main/resources/eu/dnetlib/dhp/wf/subworkflows/resulttoorganizationfrominstrepo/oozie_app/workflow.xml

@@ -5,8 +5,8 @@
             <description>the source path</description>
         </property>
         <property>
-            <name>outputPath</name>
+            <name>blacklist</name>
-            <description>sets the outputPath</description>
+            <description>The list of institutional repositories that should not be used for the propagation</description>
         </property>
     </parameters>
 
@@ -29,97 +29,13 @@
 
     <action name="reset_outputpath">
         <fs>
-            <delete path="${outputPath}"/>
+            <delete path="${workingDir}"/>
-            <mkdir path="${outputPath}"/>
+            <mkdir path="${workingDir}"/>
         </fs>
-        <ok to="copy_entities"/>
+        <ok to="prepare_result_organization_association"/>
         <error to="Kill"/>
     </action>
 
-    <fork name="copy_entities">
-        <path start="copy_relation"/>
-        <path start="copy_publication"/>
-        <path start="copy_dataset"/>
-        <path start="copy_orp"/>
-        <path start="copy_software"/>
-        <path start="copy_organization"/>
-        <path start="copy_projects"/>
-        <path start="copy_datasources"/>
-    </fork>
-
-    <action name="copy_relation">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/relation</arg>
-            <arg>${nameNode}/${outputPath}/relation</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_publication">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/publication</arg>
-            <arg>${nameNode}/${outputPath}/publication</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_dataset">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/dataset</arg>
-            <arg>${nameNode}/${outputPath}/dataset</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_orp">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/otherresearchproduct</arg>
-            <arg>${nameNode}/${outputPath}/otherresearchproduct</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_software">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/software</arg>
-            <arg>${nameNode}/${outputPath}/software</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_organization">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/organization</arg>
-            <arg>${nameNode}/${outputPath}/organization</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_projects">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/project</arg>
-            <arg>${nameNode}/${outputPath}/project</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <action name="copy_datasources">
-        <distcp xmlns="uri:oozie:distcp-action:0.2">
-            <arg>${nameNode}/${sourcePath}/datasource</arg>
-            <arg>${nameNode}/${outputPath}/datasource</arg>
-        </distcp>
-        <ok to="wait"/>
-        <error to="Kill"/>
-    </action>
-
-    <join name="wait" to="prepare_result_organization_association"/>
 
     <action name="prepare_result_organization_association">
         <spark xmlns="uri:oozie:spark-action:0.2">
@@ -138,10 +54,9 @@
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
-            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/affiliationInstRepo</arg>
-            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/preparedInfo/datasourceOrganization</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
             <arg>--blacklist</arg><arg>${blacklist}</arg>
+            <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
         </spark>
         <ok to="fork_join_apply_resulttoorganization_propagation"/>
         <error to="Kill"/>
@@ -173,9 +88,9 @@
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/publication</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/affiliationInstRepo/publication/relation</arg>
-            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/preparedInfo/datasourceOrganization</arg>
+            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/datasourceOrganization</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/alreadyLinked</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
         </spark>
@@ -202,9 +117,9 @@
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/dataset</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/affiliationInstRepo/dataset/relation</arg>
-            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/preparedInfo/datasourceOrganization</arg>
+            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/datasourceOrganization</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/alreadyLinked</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
         </spark>
@@ -231,9 +146,9 @@
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/otherresearchproduct</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/affiliationInstRepo/otherresearchproduct/relation</arg>
-            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/preparedInfo/datasourceOrganization</arg>
+            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/datasourceOrganization</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/alreadyLinked</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
         </spark>
@@ -260,9 +175,9 @@
                 --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
             </spark-opts>
             <arg>--sourcePath</arg><arg>${sourcePath}/software</arg>
-            <arg>--outputPath</arg><arg>${outputPath}/relation</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/affiliationInstRepo/software/relation</arg>
-            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/preparedInfo/datasourceOrganization</arg>
+            <arg>--datasourceOrganizationPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/datasourceOrganization</arg>
-            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/preparedInfo/alreadyLinked</arg>
+            <arg>--alreadyLinkedPath</arg><arg>${workingDir}/affiliationInstRepo/preparedInfo/alreadyLinked</arg>
             <arg>--hive_metastore_uris</arg><arg>${hive_metastore_uris}</arg>
             <arg>--resultTableName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
         </spark>
@@ -270,7 +185,32 @@
         <error to="Kill"/>
     </action>
 
-    <join name="wait2" to="End"/>
+    <join name="wait2" to="append_new_relations"/>
+
+    <action name="append_new_relations">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>append new relations</name>
+            <class>eu.dnetlib.dhp.resulttoorganizationfrominstrepo.AppendNewRelations</class>
+            <jar>dhp-enrichment-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-cores=${sparkExecutorCores}
+                --executor-memory=${sparkExecutorMemory}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.dynamicAllocation.enabled=true
+                --conf spark.dynamicAllocation.maxExecutors=${spark2MaxExecutors}
+            </spark-opts>
+            <arg>--outputPath</arg><arg>${sourcePath}/relation</arg>
+            <arg>--sourcePath</arg><arg>${workingDir}/affiliationInstRepo/</arg>
+        </spark>
+        <ok to="End"/>
+        <error to="Kill"/>
+    </action>
 
     <end name="End"/>
 

dhp-workflows/dhp-enrichment/src/test/java/eu/dnetlib/dhp/entitytoorganizationfromsemrel/SparkJobTest.java

@@ -114,7 +114,7 @@ public class SparkJobTest {
 			.option("compression", "gzip")
 			.json(workingDir.toString() + "/projectInput");
 
-		SparkResultToOrganizationFromSemRel
+		SparkEntityToOrganizationFromSemRel
 
 			.main(
 				new String[] {
@@ -395,7 +395,7 @@ public class SparkJobTest {
 			.option("compression", "gzip")
 			.json(workingDir.toString() + "/projectInput");
 
-		SparkResultToOrganizationFromSemRel
+		SparkEntityToOrganizationFromSemRel
 
 			.main(
 				new String[] {
@@ -678,7 +678,7 @@ public class SparkJobTest {
 			.option("compression", "gzip")
 			.json(workingDir.toString() + "/projectInput");
 
-		SparkResultToOrganizationFromSemRel
+		SparkEntityToOrganizationFromSemRel
 
 			.main(
 				new String[] {

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/oozie_app/download.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 curl -LSs $1 | hdfs dfs -put - $2/$3
-curl -LSs  http://api.crossref.org/works/10.1099/jgv.0.001453 > prova.txt
+#curl -LSs  http://api.crossref.org/works/10.1099/jgv.0.001453 > prova.txt

dhp-workflows/dhp-graph-provision/src/test/java/eu/dnetlib/dhp/oa/provision/IndexRecordTransformerTest.java

@@ -82,7 +82,7 @@ public class IndexRecordTransformerTest {
 	void testPeerReviewed() throws IOException, TransformerException {
 
 		final XmlRecordFactory xmlRecordFactory = new XmlRecordFactory(contextMapper, false,
-				XmlConverterJob.schemaLocation);
+			XmlConverterJob.schemaLocation);
 
 		final Publication p = load("publication.json", Publication.class);
 

dhp-workflows/dhp-graph-provision/src/test/java/eu/dnetlib/dhp/oa/provision/XmlIndexingJobTest.java

@@ -1,6 +1,8 @@
 
 package eu.dnetlib.dhp.oa.provision;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 import java.io.IOException;
 import java.io.StringReader;
 import java.net.URI;
@@ -32,8 +34,6 @@ import eu.dnetlib.dhp.oa.provision.utils.ISLookupClient;
 import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
 import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
 @ExtendWith(MockitoExtension.class)
 public class XmlIndexingJobTest extends SolrTest {
 
@@ -110,34 +110,33 @@ public class XmlIndexingJobTest extends SolrTest {
 		QueryResponse rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "*:*"));
 
 		assertEquals(
-				nRecord, rsp.getResults().getNumFound(),
+			nRecord, rsp.getResults().getNumFound(),
-				"the number of indexed records should be equal to the number of input records");
+			"the number of indexed records should be equal to the number of input records");
-
 
 		rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "isgreen:true"));
 		assertEquals(
-				0, rsp.getResults().getNumFound(),
+			0, rsp.getResults().getNumFound(),
-				"the number of indexed records having isgreen = true");
+			"the number of indexed records having isgreen = true");
 
 		rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "openaccesscolor:bronze"));
 		assertEquals(
-				0, rsp.getResults().getNumFound(),
+			0, rsp.getResults().getNumFound(),
-				"the number of indexed records having openaccesscolor = bronze");
+			"the number of indexed records having openaccesscolor = bronze");
 
 		rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "isindiamondjournal:true"));
 		assertEquals(
-				0, rsp.getResults().getNumFound(),
+			0, rsp.getResults().getNumFound(),
-				"the number of indexed records having isindiamondjournal = true");
+			"the number of indexed records having isindiamondjournal = true");
 
 		rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "publiclyfunded:true"));
 		assertEquals(
-				0, rsp.getResults().getNumFound(),
+			0, rsp.getResults().getNumFound(),
-				"the number of indexed records having publiclyfunded = true");
+			"the number of indexed records having publiclyfunded = true");
 
 		rsp = miniCluster.getSolrClient().query(new SolrQuery().add(CommonParams.Q, "peerreviewed:true"));
 		assertEquals(
-				0, rsp.getResults().getNumFound(),
+			0, rsp.getResults().getNumFound(),
-				"the number of indexed records having peerreviewed = true");
+			"the number of indexed records having peerreviewed = true");
 	}
 
 	@Test