Merge pull request 'cleanCountryOnMaster' (#265) from cleanCountryOnMaster into master

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

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/country/CleanCountrySparkJob.java

@@ -0,0 +1,211 @@
+
+package eu.dnetlib.dhp.oa.graph.clean.country;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import javax.swing.text.html.Option;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FilterFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author miriam.baglioni
+ * @Date 20/07/22
+ */
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob;
+import eu.dnetlib.dhp.schema.oaf.Country;
+import eu.dnetlib.dhp.schema.oaf.Instance;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import eu.dnetlib.dhp.schema.oaf.utils.PidType;
+
+public class CleanCountrySparkJob implements Serializable {
+    private static final Logger log = LoggerFactory.getLogger(CleanCountrySparkJob.class);
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    public static void main(String[] args) throws Exception {
+
+        String jsonConfiguration = IOUtils
+                .toString(
+                        CleanCountrySparkJob.class
+                                .getResourceAsStream(
+                                        "/eu/dnetlib/dhp/oa/graph/input_clean_country_parameters.json"));
+        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+        parser.parseArgument(args);
+
+        Boolean isSparkSessionManaged = Optional
+                .ofNullable(parser.get("isSparkSessionManaged"))
+                .map(Boolean::valueOf)
+                .orElse(Boolean.TRUE);
+        log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+        String inputPath = parser.get("inputPath");
+        log.info("inputPath: {}", inputPath);
+
+        String workingDir = parser.get("workingDir");
+        log.info("workingDir: {}", workingDir);
+
+        String datasourcePath = parser.get("hostedBy");
+        log.info("datasourcePath: {}", datasourcePath);
+
+        String country = parser.get("country");
+        log.info("country: {}", country);
+
+        String[] verifyParam = parser.get("verifyParam").split(";");
+        log.info("verifyParam: {}", verifyParam);
+
+        String collectedfrom = parser.get("collectedfrom");
+        log.info("collectedfrom: {}", collectedfrom);
+
+        String graphTableClassName = parser.get("graphTableClassName");
+        log.info("graphTableClassName: {}", graphTableClassName);
+
+        Class<? extends Result> entityClazz = (Class<? extends Result>) Class.forName(graphTableClassName);
+
+        SparkConf conf = new SparkConf();
+        runWithSparkSession(
+                conf,
+                isSparkSessionManaged,
+                spark -> {
+
+                    cleanCountry(
+                            spark, country, verifyParam, inputPath, entityClazz, workingDir, collectedfrom, datasourcePath);
+                });
+    }
+
+    private static <T extends Result> void cleanCountry(SparkSession spark, String country, String[] verifyParam,
+                                                        String inputPath, Class<T> entityClazz, String workingDir, String collectedfrom, String datasourcePath) {
+
+        List<String> hostedBy = spark
+                .read()
+                .textFile(datasourcePath)
+                .collectAsList();
+
+        Dataset<T> res = spark
+                .read()
+                .textFile(inputPath)
+                .map(
+                        (MapFunction<String, T>) value -> OBJECT_MAPPER.readValue(value, entityClazz),
+                        Encoders.bean(entityClazz));
+
+        res.map((MapFunction<T, T>) r -> {
+                    if (r.getInstance().stream().anyMatch(i -> hostedBy.contains(i.getHostedby().getKey())) ||
+                            !r.getCollectedfrom().stream().anyMatch(cf -> cf.getValue().equals(collectedfrom))) {
+                        return r;
+                    }
+
+                    List<StructuredProperty> ids = getPidsAndAltIds(r).collect(Collectors.toList());
+                    if (ids
+                            .stream()
+                            .anyMatch(
+                                    p -> p
+                                            .getQualifier()
+                                            .getClassid()
+                                            .equals(PidType.doi.toString()) && pidInParam(p.getValue(), verifyParam))) {
+                        r
+                                .setCountry(
+                                        r
+                                                .getCountry()
+                                                .stream()
+                                                .filter(
+                                                        c -> toTakeCountry(c, country))
+                                                .collect(Collectors.toList()));
+
+                    }
+
+                    return r;
+                }, Encoders.bean(entityClazz))
+                .write()
+                .mode(SaveMode.Overwrite)
+                .option("compression", "gzip")
+                .json(workingDir);
+
+        spark
+                .read()
+                .textFile(workingDir)
+                .map(
+                        (MapFunction<String, T>) value -> OBJECT_MAPPER.readValue(value, entityClazz),
+                        Encoders.bean(entityClazz))
+                .write()
+                .mode(SaveMode.Overwrite)
+                .option("compression", "gzip")
+                .json(inputPath);
+    }
+
+    private static <T extends Result> Stream<StructuredProperty> getPidsAndAltIds(T r) {
+        final Stream<StructuredProperty> resultPids = Optional
+                .ofNullable(r.getPid())
+                .map(Collection::stream)
+                .orElse(Stream.empty());
+
+        final Stream<StructuredProperty> instancePids = Optional
+                .ofNullable(r.getInstance())
+                .map(
+                        instance -> instance
+                                .stream()
+                                .flatMap(
+                                        i -> Optional
+                                                .ofNullable(i.getPid())
+                                                .map(Collection::stream)
+                                                .orElse(Stream.empty())))
+                .orElse(Stream.empty());
+
+        final Stream<StructuredProperty> instanceAltIds = Optional
+                .ofNullable(r.getInstance())
+                .map(
+                        instance -> instance
+                                .stream()
+                                .flatMap(
+                                        i -> Optional
+                                                .ofNullable(i.getAlternateIdentifier())
+                                                .map(Collection::stream)
+                                                .orElse(Stream.empty())))
+                .orElse(Stream.empty());
+
+        return Stream
+                .concat(
+                        Stream.concat(resultPids, instancePids),
+                        instanceAltIds);
+    }
+
+    private static boolean pidInParam(String value, String[] verifyParam) {
+        for (String s : verifyParam)
+            if (value.startsWith(s))
+                return true;
+        return false;
+    }
+
+    private static boolean toTakeCountry(Country c, String country) {
+        // If dataInfo is not set, or dataInfo.inferenceprovenance is not set or not present then it cannot be
+        // inserted via propagation
+        if (!Optional.ofNullable(c.getDataInfo()).isPresent())
+            return true;
+        if (!Optional.ofNullable(c.getDataInfo().getInferenceprovenance()).isPresent())
+            return true;
+        return !(c
+                .getClassid()
+                .equalsIgnoreCase(country) &&
+                c.getDataInfo().getInferenceprovenance().equals("propagation"));
+    }
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/country/GetDatasourceFromCountry.java

@@ -0,0 +1,106 @@
+
+package eu.dnetlib.dhp.oa.graph.clean.country;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FilterFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.oa.graph.clean.CleanContextSparkJob;
+import eu.dnetlib.dhp.schema.common.ModelConstants;
+import eu.dnetlib.dhp.schema.oaf.*;
+import scala.Tuple2;
+
+/**
+ * @author miriam.baglioni
+ * @Date 22/07/22
+ */
+public class GetDatasourceFromCountry implements Serializable {
+    private static final Logger log = LoggerFactory.getLogger(GetDatasourceFromCountry.class);
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    public static void main(String[] args) throws Exception {
+
+        String jsonConfiguration = IOUtils
+                .toString(
+                        GetDatasourceFromCountry.class
+                                .getResourceAsStream(
+                                        "/eu/dnetlib/dhp/oa/graph/input_datasource_country_parameters.json"));
+        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+        parser.parseArgument(args);
+
+        Boolean isSparkSessionManaged = Optional
+                .ofNullable(parser.get("isSparkSessionManaged"))
+                .map(Boolean::valueOf)
+                .orElse(Boolean.TRUE);
+        log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+        String inputPath = parser.get("inputPath");
+        log.info("inputPath: {}", inputPath);
+
+        String workingPath = parser.get("workingDir");
+        log.info("workingDir: {}", workingPath);
+
+        String country = parser.get("country");
+        log.info("country: {}", country);
+
+        SparkConf conf = new SparkConf();
+        runWithSparkSession(
+                conf,
+                isSparkSessionManaged,
+                spark -> {
+                    getDatasourceFromCountry(spark, country, inputPath, workingPath);
+                });
+    }
+
+    private static void getDatasourceFromCountry(SparkSession spark, String country, String inputPath,
+                                                 String workingDir) {
+
+        Dataset<Organization> organization = spark
+                .read()
+                .textFile(inputPath + "/organization")
+                .map(
+                        (MapFunction<String, Organization>) value -> OBJECT_MAPPER.readValue(value, Organization.class),
+                        Encoders.bean(Organization.class))
+                .filter(
+                        (FilterFunction<Organization>) o -> !o.getDataInfo().getDeletedbyinference() &&
+                                o.getCountry().getClassid().length() > 0 &&
+                                o.getCountry().getClassid().equals(country));
+
+        // filtering of the relations taking the non deleted by inference and those with IsProvidedBy as relclass
+        Dataset<Relation> relation = spark
+                .read()
+                .textFile(inputPath + "/relation")
+                .map(
+                        (MapFunction<String, Relation>) value -> OBJECT_MAPPER.readValue(value, Relation.class),
+                        Encoders.bean(Relation.class))
+                .filter(
+                        (FilterFunction<Relation>) rel -> rel.getRelClass().equalsIgnoreCase(ModelConstants.IS_PROVIDED_BY) &&
+                                !rel.getDataInfo().getDeletedbyinference());
+
+        organization
+                .joinWith(relation, organization.col("id").equalTo(relation.col("target")))
+                .map((MapFunction<Tuple2<Organization, Relation>, String>) t2 -> t2._2().getSource(), Encoders.STRING())
+                .write()
+                .mode(SaveMode.Overwrite)
+                .option("compression", "gzip")
+                .json(workingDir);
+
+    }
+}

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/clean/oozie_app/workflow.xml

@@ -14,8 +14,8 @@
             <description>the address of the lookUp service</description>
         </property>
         <property>
-            <name>shouldCleanContext</name>
+            <name>shouldClean</name>
-            <description>true if the context have to be cleaned</description>
+            <description>true if the operation of deletion of not needed values from the results have to be performed</description>
         </property>
         <property>
             <name>contextId</name>
@@ -30,6 +30,22 @@
             <description>It is the constrint to be verified. This time is hardcoded as gcube and it is searched for in
             the title. If title starts with gcube than the context sobigdata will be removed by the result if present</description>
         </property>
+        <property>
+            <name>verifyCountryParam</name>
+            <value>10.17632;10.5061</value>
+            <description>It is the constraints to be verified. This time is hardcoded as the starting doi from mendeley and dryad and it is searched for in
+                the pid value. If the pid value starts with one of the two prefixes, then the country may be removed</description>
+        </property>
+        <property>
+            <name>country</name>
+            <value>NL</value>
+            <description>It is the country to be removed from the set of countries if it is present with provenance propagation. The country will not be removed if in one of the isntances there is a datasource with country `country`</description>
+        </property>
+        <property>
+            <name>collectedfrom</name>
+            <value>NARCIS</value>
+            <description>the only datasource for which the country NL will be removed from the country list</description>
+        </property>
 
         <property>
             <name>sparkDriverMemory</name>
@@ -296,18 +312,18 @@
 
     <decision name="clean_context">
         <switch>
-            <case to="fork_clean_context">${wf:conf('shouldCleanContext') eq true}</case>
+            <case to="fork_clean_context">${wf:conf('shouldClean') eq true}</case>
             <default to="End"/>
         </switch>
     </decision>
 
-
     <fork name="fork_clean_context">
         <path start="clean_publication_context"/>
         <path start="clean_dataset_context"/>
         <path start="clean_otherresearchproduct_context"/>
         <path start="clean_software_context"/>
     </fork>
+
     <action name="clean_publication_context">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>
@@ -327,7 +343,7 @@
             </spark-opts>
             <arg>--inputPath</arg><arg>${graphOutputPath}/publication</arg>
             <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
-            <arg>--workingPath</arg><arg>${workingDir}/working/publication</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/publication</arg>
             <arg>--contextId</arg><arg>${contextId}</arg>
             <arg>--verifyParam</arg><arg>${verifyParam}</arg>
         </spark>
@@ -354,7 +370,7 @@
             </spark-opts>
             <arg>--inputPath</arg><arg>${graphOutputPath}/dataset</arg>
             <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
-            <arg>--workingPath</arg><arg>${workingDir}/working/dataset</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/dataset</arg>
             <arg>--contextId</arg><arg>${contextId}</arg>
             <arg>--verifyParam</arg><arg>${verifyParam}</arg>
         </spark>
@@ -381,7 +397,7 @@
             </spark-opts>
             <arg>--inputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
             <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
-            <arg>--workingPath</arg><arg>${workingDir}/working/otherresearchproduct</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/otherresearchproduct</arg>
             <arg>--contextId</arg><arg>${contextId}</arg>
             <arg>--verifyParam</arg><arg>${verifyParam}</arg>
         </spark>
@@ -408,7 +424,7 @@
             </spark-opts>
             <arg>--inputPath</arg><arg>${graphOutputPath}/software</arg>
             <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
-            <arg>--workingPath</arg><arg>${workingDir}/working/software</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/software</arg>
             <arg>--contextId</arg><arg>${contextId}</arg>
             <arg>--verifyParam</arg><arg>${verifyParam}</arg>
         </spark>
@@ -416,7 +432,352 @@
         <error to="Kill"/>
     </action>
 
-    <join name="wait_clean_context" to="End"/>
+    <join name="wait_clean_context" to="select_datasourceId_from_country"/>
+
+    <action name="select_datasourceId_from_country">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Select datasource ID from country</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.country.GetDatasourceFromCountry</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/hostedby</arg>
+            <arg>--country</arg><arg>${country}</arg>
+        </spark>
+        <ok to="fork_clean_country"/>
+        <error to="Kill"/>
+    </action>
+
+    <fork name="fork_clean_country">
+        <path start="clean_publication_country"/>
+        <path start="clean_dataset_country"/>
+        <path start="clean_otherresearchproduct_country"/>
+        <path start="clean_software_country"/>
+    </fork>
+
+    <action name="clean_publication_country">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean publication country</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.country.CleanCountrySparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/publication</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/publication</arg>
+            <arg>--country</arg><arg>${country}</arg>
+            <arg>--verifyParam</arg><arg>${verifyCountryParam}</arg>
+            <arg>--hostedBy</arg><arg>${workingDir}/working/hostedby</arg>
+            <arg>--collectedfrom</arg><arg>${collectedfrom}</arg>
+        </spark>
+        <ok to="wait_clean_country"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_dataset_country">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean dataset country</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.country.CleanCountrySparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/dataset</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/dataset</arg>
+            <arg>--country</arg><arg>${country}</arg>
+            <arg>--verifyParam</arg><arg>${verifyCountryParam}</arg>
+            <arg>--hostedBy</arg><arg>${workingDir}/working/hostedby</arg>
+            <arg>--collectedfrom</arg><arg>${collectedfrom}</arg>
+        </spark>
+        <ok to="wait_clean_country"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_otherresearchproduct_country">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean otherresearchproduct country</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.country.CleanCountrySparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/otherresearchproduct</arg>
+            <arg>--country</arg><arg>${country}</arg>
+            <arg>--verifyParam</arg><arg>${verifyCountryParam}</arg>
+            <arg>--hostedBy</arg><arg>${workingDir}/working/hostedby</arg>
+            <arg>--collectedfrom</arg><arg>${collectedfrom}</arg>
+        </spark>
+        <ok to="wait_clean_country"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="clean_software_country">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>Clean software country</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.country.CleanCountrySparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/software</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
+            <arg>--workingDir</arg><arg>${workingDir}/working/software</arg>
+            <arg>--country</arg><arg>${country}</arg>
+            <arg>--verifyParam</arg><arg>${verifyCountryParam}</arg>
+            <arg>--hostedBy</arg><arg>${workingDir}/working/hostedby</arg>
+            <arg>--collectedfrom</arg><arg>${collectedfrom}</arg>
+        </spark>
+        <ok to="wait_clean_country"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="wait_clean_country" to="should_patch_datasource_ids"/>
+
+    <decision name="should_patch_datasource_ids">
+        <switch>
+            <case to="get_ds_master_duplicate">${wf:conf('shouldClean') eq true}</case>
+            <default to="End"/>
+        </switch>
+    </decision>
+
+    <action name="get_ds_master_duplicate">
+        <java>
+            <main-class>eu.dnetlib.dhp.oa.graph.clean.MasterDuplicateAction</main-class>
+            <arg>--postgresUrl</arg><arg>${postgresURL}</arg>
+            <arg>--postgresUser</arg><arg>${postgresUser}</arg>
+            <arg>--postgresPassword</arg><arg>${postgresPassword}</arg>
+            <arg>--hdfsPath</arg><arg>${workingDir}/masterduplicate</arg>
+            <arg>--hdfsNameNode</arg><arg>${nameNode}</arg>
+        </java>
+        <ok to="fork_patch_cfhb"/>
+        <error to="Kill"/>
+    </action>
+
+    <fork name="fork_patch_cfhb">
+        <path start="patch_publication_cfhb"/>
+        <path start="patch_dataset_cfhb"/>
+        <path start="patch_otherresearchproduct_cfhb"/>
+        <path start="patch_software_cfhb"/>
+    </fork>
+
+    <action name="patch_publication_cfhb">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>patch publication cfhb</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.cfhb.CleanCfHbSparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/publication</arg>
+            <arg>--resolvedPath</arg><arg>${workingDir}/cfHbResolved/publication</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/cfHbPatched/publication</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Publication</arg>
+            <arg>--masterDuplicatePath</arg><arg>${workingDir}/masterduplicate</arg>
+        </spark>
+        <ok to="wait_clean_cfhb"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="patch_dataset_cfhb">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>patch dataset cfhb</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.cfhb.CleanCfHbSparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/dataset</arg>
+            <arg>--resolvedPath</arg><arg>${workingDir}/cfHbResolved/dataset</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/cfHbPatched/dataset</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Dataset</arg>
+            <arg>--masterDuplicatePath</arg><arg>${workingDir}/masterduplicate</arg>
+        </spark>
+        <ok to="wait_clean_cfhb"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="patch_otherresearchproduct_cfhb">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>patch otherresearchproduct cfhb</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.cfhb.CleanCfHbSparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/otherresearchproduct</arg>
+            <arg>--resolvedPath</arg><arg>${workingDir}/cfHbResolved/otherresearchproduct</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/cfHbPatched/otherresearchproduct</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.OtherResearchProduct</arg>
+            <arg>--masterDuplicatePath</arg><arg>${workingDir}/masterduplicate</arg>
+        </spark>
+        <ok to="wait_clean_cfhb"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="patch_software_cfhb">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn</master>
+            <mode>cluster</mode>
+            <name>patch software cfhb</name>
+            <class>eu.dnetlib.dhp.oa.graph.clean.cfhb.CleanCfHbSparkJob</class>
+            <jar>dhp-graph-mapper-${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.sql.shuffle.partitions=7680
+            </spark-opts>
+            <arg>--inputPath</arg><arg>${graphOutputPath}/software</arg>
+            <arg>--resolvedPath</arg><arg>${workingDir}/cfHbResolved/software</arg>
+            <arg>--outputPath</arg><arg>${workingDir}/cfHbPatched/software</arg>
+            <arg>--graphTableClassName</arg><arg>eu.dnetlib.dhp.schema.oaf.Software</arg>
+            <arg>--masterDuplicatePath</arg><arg>${workingDir}/masterduplicate</arg>
+        </spark>
+        <ok to="wait_clean_cfhb"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="wait_clean_cfhb" to="fork_copy_cfhb_patched_results"/>
+
+    <fork name="fork_copy_cfhb_patched_results">
+        <path start="copy_cfhb_patched_publication"/>
+        <path start="copy_cfhb_patched_dataset"/>
+        <path start="copy_cfhb_patched_otherresearchproduct"/>
+        <path start="copy_cfhb_patched_software"/>
+    </fork>
+
+    <action name="copy_cfhb_patched_publication">
+        <distcp xmlns="uri:oozie:distcp-action:0.2">
+            <prepare>
+                <delete path="${graphOutputPath}/publication"/>
+            </prepare>
+            <arg>${workingDir}/cfHbPatched/publication</arg>
+            <arg>${graphOutputPath}/publication</arg>
+        </distcp>
+        <ok to="copy_wait"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="copy_cfhb_patched_dataset">
+        <distcp xmlns="uri:oozie:distcp-action:0.2">
+            <prepare>
+                <delete path="${graphOutputPath}/dataset"/>
+            </prepare>
+            <arg>${workingDir}/cfHbPatched/dataset</arg>
+            <arg>${graphOutputPath}/dataset</arg>
+        </distcp>
+        <ok to="copy_wait"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="copy_cfhb_patched_otherresearchproduct">
+        <distcp xmlns="uri:oozie:distcp-action:0.2">
+            <prepare>
+                <delete path="${graphOutputPath}/otherresearchproduct"/>
+            </prepare>
+            <arg>${workingDir}/cfHbPatched/otherresearchproduct</arg>
+            <arg>${graphOutputPath}/otherresearchproduct</arg>
+        </distcp>
+        <ok to="copy_wait"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="copy_cfhb_patched_software">
+        <distcp xmlns="uri:oozie:distcp-action:0.2">
+            <prepare>
+                <delete path="${graphOutputPath}/software"/>
+            </prepare>
+            <arg>${workingDir}/cfHbPatched/software</arg>
+            <arg>${graphOutputPath}/software</arg>
+        </distcp>
+        <ok to="copy_wait"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="copy_wait" to="End"/>
 
     <end name="End"/>
+
 </workflow-app>

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/input_clean_country_parameters.json

@@ -0,0 +1,49 @@
+[
+  {
+    "paramName": "issm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "when true will stop SparkSession after job execution",
+    "paramRequired": false
+  },
+  {
+    "paramName": "in",
+    "paramLongName": "inputPath",
+    "paramDescription": "the path to the graph data dump to read",
+    "paramRequired": true
+  },
+  {
+    "paramName": "wd",
+    "paramLongName": "workingDir",
+    "paramDescription": "the path to store the output graph",
+    "paramRequired": true
+  },
+  {
+    "paramName": "c",
+    "paramLongName": "country",
+    "paramDescription": "the id of the context to be removed",
+    "paramRequired": true
+  },
+  {
+    "paramName": "class",
+    "paramLongName": "graphTableClassName",
+    "paramDescription": "class name moelling the graph table",
+    "paramRequired": true
+  },{
+  "paramName": "vf",
+  "paramLongName": "verifyParam",
+  "paramDescription": "the parameter to be verified to remove the country",
+  "paramRequired": true
+},
+  {
+    "paramName": "cf",
+    "paramLongName": "collectedfrom",
+    "paramDescription": "the collectedfrom value for which we should apply the cleaning",
+    "paramRequired": true
+  },
+  {
+    "paramName": "hb",
+    "paramLongName": "hostedBy",
+    "paramDescription": "the set of datasources having the specified country in the graph searched for in the hostedby of the results",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/clean/CleanCountryTest.java

@@ -0,0 +1,191 @@
+
+package eu.dnetlib.dhp.oa.graph.clean;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import eu.dnetlib.dhp.schema.oaf.Dataset;
+import org.apache.commons.io.FileUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author miriam.baglioni
+ * @Date 20/07/22
+ */
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.oa.graph.clean.country.CleanCountrySparkJob;
+import eu.dnetlib.dhp.schema.oaf.Publication;
+
+public class CleanCountryTest {
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    private static SparkSession spark;
+
+    private static Path workingDir;
+
+    private static final Logger log = LoggerFactory.getLogger(CleanContextTest.class);
+
+    @BeforeAll
+    public static void beforeAll() throws IOException {
+        workingDir = Files.createTempDirectory(CleanCountryTest.class.getSimpleName());
+        log.info("using work dir {}", workingDir);
+
+        SparkConf conf = new SparkConf();
+        conf.setAppName(CleanCountryTest.class.getSimpleName());
+
+        conf.setMaster("local[*]");
+        conf.set("spark.driver.host", "localhost");
+        conf.set("hive.metastore.local", "true");
+        conf.set("spark.ui.enabled", "false");
+        conf.set("spark.sql.warehouse.dir", workingDir.toString());
+        conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+        spark = SparkSession
+                .builder()
+                .appName(CleanCountryTest.class.getSimpleName())
+                .config(conf)
+                .getOrCreate();
+    }
+
+    @AfterAll
+    public static void afterAll() throws IOException {
+        FileUtils.deleteDirectory(workingDir.toFile());
+        spark.stop();
+    }
+
+    @Test
+    public void testResultClean() throws Exception {
+        final String sourcePath = getClass()
+                .getResource("/eu/dnetlib/dhp/oa/graph/clean/publication_clean_country.json")
+                .getPath();
+
+        spark
+                .read()
+                .textFile(sourcePath)
+                .map(
+                        (MapFunction<String, Publication>) r -> OBJECT_MAPPER.readValue(r, Publication.class),
+                        Encoders.bean(Publication.class))
+                .write()
+                .json(workingDir.toString() + "/publication");
+
+        CleanCountrySparkJob.main(new String[] {
+                "--isSparkSessionManaged", Boolean.FALSE.toString(),
+                "--inputPath", workingDir.toString() + "/publication",
+                "--graphTableClassName", Publication.class.getCanonicalName(),
+                "--workingDir", workingDir.toString() + "/working",
+                "--country", "NL",
+                "--verifyParam", "10.17632",
+                "--collectedfrom", "NARCIS",
+                "--hostedBy", getClass()
+                .getResource("/eu/dnetlib/dhp/oa/graph/clean/hostedBy")
+                .getPath()
+        });
+
+        final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+        JavaRDD<Publication> tmp = sc
+                .textFile(workingDir.toString() + "/publication")
+                .map(item -> OBJECT_MAPPER.readValue(item, Publication.class));
+
+        Assertions.assertEquals(8, tmp.count());
+
+        // original result with NL country and doi starting with Mendely prefix, but not collectedfrom NARCIS
+        Assertions
+                .assertEquals(
+                        1,
+                        tmp
+                                .filter(p -> p.getId().equals("50|DansKnawCris::0224aae28af558f21768dbc6439c7a95"))
+                                .collect()
+                                .get(0)
+                                .getCountry()
+                                .size());
+
+        // original result with NL country and pid not starting with Mendely prefix
+        Assertions
+                .assertEquals(
+                        1,
+                        tmp
+                                .filter(p -> p.getId().equals("50|DansKnawCris::20c414a3b1c742d5dd3851f1b67df2d9"))
+                                .collect()
+                                .get(0)
+                                .getCountry()
+                                .size());
+
+        // original result with NL country and doi starting with Mendely prefix and collectedfrom NARCIS but not
+        // inserted with propagation
+        Assertions
+                .assertEquals(
+                        1,
+                        tmp
+                                .filter(p -> p.getId().equals("50|DansKnawCris::3c81248c335f0aa07e06817ece6fa6af"))
+                                .collect()
+                                .get(0)
+                                .getCountry()
+                                .size());
+
+        // original result with NL country and doi starting with Mendely prefix and collectedfrom NARCIS inserted with
+        // propagation
+        Assertions
+                .assertEquals(
+                        0,
+                        tmp
+                                .filter(p -> p.getId().equals("50|DansKnawCris::3c81248c335f0aa07e06817ece6fa6ag"))
+                                .collect()
+                                .get(0)
+                                .getCountry()
+                                .size());
+    }
+
+    @Test
+    public void testDatasetClean() throws Exception {
+        final String sourcePath = getClass()
+                .getResource("/eu/dnetlib/dhp/oa/graph/clean/dataset_clean_country.json")
+                .getPath();
+
+        spark
+                .read()
+                .textFile(sourcePath)
+                .map(
+                        (MapFunction<String, Dataset>) r -> OBJECT_MAPPER.readValue(r, Dataset.class),
+                        Encoders.bean(Dataset.class))
+                .write()
+                .json(workingDir.toString() + "/dataset");
+
+        CleanCountrySparkJob.main(new String[] {
+                "--isSparkSessionManaged", Boolean.FALSE.toString(),
+                "--inputPath", workingDir.toString() + "/dataset",
+                "-graphTableClassName", Dataset.class.getCanonicalName(),
+                "-workingDir", workingDir.toString() + "/working",
+                "-country", "NL",
+                "-verifyParam", "10.17632",
+                "-collectedfrom", "NARCIS",
+                "-hostedBy", getClass()
+                .getResource("/eu/dnetlib/dhp/oa/graph/clean/hostedBy")
+                .getPath()
+        });
+
+        final JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
+        JavaRDD<Dataset> tmp = sc
+                .textFile(workingDir.toString() + "/dataset")
+                .map(item -> OBJECT_MAPPER.readValue(item, Dataset.class));
+
+        Assertions.assertEquals(1, tmp.count());
+
+        Assertions.assertEquals(0, tmp.first().getCountry().size());
+
+
+    }
+
+}