refactoring

dhp-workflows/dhp-bulktag/project-default.properties

@@ -0,0 +1,7 @@
+#sandboxName when not provided explicitly will be generated
+sandboxName=${sandboxName}
+sandboxDir=/user/${dhp.hadoop.frontend.user.name}/${sandboxName}
+workingDir=${sandboxDir}/working_dir
+oozie.wf.application.path = ${nameNode}${sandboxDir}/${oozieAppDir}
+oozieTopWfApplicationPath = ${oozie.wf.application.path}
+

dhp-workflows/dhp-bulktag/src/main/java/eu/dnetlib/dhp/bulktag/SparkBulkTagJob2.java

@@ -1,4 +1,4 @@
-package eu.dnetlib.dhp;
+package eu.dnetlib.dhp.bulktag;
 
 import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 
@@ -28,7 +28,7 @@ public class SparkBulkTagJob2 {
         String jsonConfiguration =
                 IOUtils.toString(
                         SparkBulkTagJob2.class.getResourceAsStream(
-                                "/eu/dnetlib/dhp/bulktag/input_bulktag_parameters.json"));
+                                "/eu/dnetlib/dhp/bulktag/input_bulkTag_parameters.json"));
 
         final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
 

dhp-workflows/dhp-bulktag/src/main/java/eu/dnetlib/dhp/community/ResultTagger.java

@@ -65,13 +65,16 @@ public class ResultTagger implements Serializable {
 
         // tagging for Subject
         final Set<String> subjects = new HashSet<>();
-        result.getSubject().stream()
-                .map(subject -> subject.getValue())
-                .filter(StringUtils::isNotBlank)
-                .map(String::toLowerCase)
-                .map(String::trim)
-                .collect(Collectors.toCollection(HashSet::new))
-                .forEach(s -> subjects.addAll(conf.getCommunityForSubjectValue(s)));
+        Optional<List<StructuredProperty>> oresultsubj = Optional.ofNullable(result.getSubject());
+        if (oresultsubj.isPresent()) {
+            oresultsubj.get().stream()
+                    .map(subject -> subject.getValue())
+                    .filter(StringUtils::isNotBlank)
+                    .map(String::toLowerCase)
+                    .map(String::trim)
+                    .collect(Collectors.toCollection(HashSet::new))
+                    .forEach(s -> subjects.addAll(conf.getCommunityForSubjectValue(s)));
+        }
 
         communities.addAll(subjects);
 
@@ -79,32 +82,43 @@ public class ResultTagger implements Serializable {
         final Set<String> datasources = new HashSet<>();
         final Set<String> tmp = new HashSet<>();
 
-        for (Instance i : result.getInstance()) {
-            tmp.add(StringUtils.substringAfter(i.getCollectedfrom().getKey(), "|"));
-            tmp.add(StringUtils.substringAfter(i.getHostedby().getKey(), "|"));
-        }
+        Optional<List<Instance>> oresultinstance = Optional.ofNullable(result.getInstance());
+        if (oresultinstance.isPresent()) {
+            for (Instance i : oresultinstance.get()) {
+                tmp.add(StringUtils.substringAfter(i.getCollectedfrom().getKey(), "|"));
+                tmp.add(StringUtils.substringAfter(i.getHostedby().getKey(), "|"));
+            }
 
-        result.getInstance().stream()
-                .map(i -> new Pair<>(i.getCollectedfrom().getKey(), i.getHostedby().getKey()))
-                .flatMap(p -> Stream.of(p.getFst(), p.getSnd()))
-                .map(s -> StringUtils.substringAfter(s, "|"))
-                .collect(Collectors.toCollection(HashSet::new))
-                .forEach(dsId -> datasources.addAll(conf.getCommunityForDatasource(dsId, param)));
+            oresultinstance.get().stream()
+                    .map(i -> new Pair<>(i.getCollectedfrom().getKey(), i.getHostedby().getKey()))
+                    .flatMap(p -> Stream.of(p.getFst(), p.getSnd()))
+                    .map(s -> StringUtils.substringAfter(s, "|"))
+                    .collect(Collectors.toCollection(HashSet::new))
+                    .forEach(
+                            dsId ->
+                                    datasources.addAll(
+                                            conf.getCommunityForDatasource(dsId, param)));
+        }
 
         communities.addAll(datasources);
 
         /*Tagging for Zenodo Communities*/
         final Set<String> czenodo = new HashSet<>();
-        result.getContext().stream()
-                .filter(c -> c.getId().contains(ZENODO_COMMUNITY_INDICATOR))
-                .collect(Collectors.toList())
-                .forEach(
-                        c ->
-                                czenodo.addAll(
-                                        conf.getCommunityForZenodoCommunityValue(
-                                                c.getId()
-                                                        .substring(c.getId().lastIndexOf("/") + 1)
-                                                        .trim())));
+
+        Optional<List<Context>> oresultcontext = Optional.ofNullable(result.getContext());
+        if (oresultcontext.isPresent()) {
+            oresultcontext.get().stream()
+                    .filter(c -> c.getId().contains(ZENODO_COMMUNITY_INDICATOR))
+                    .collect(Collectors.toList())
+                    .forEach(
+                            c ->
+                                    czenodo.addAll(
+                                            conf.getCommunityForZenodoCommunityValue(
+                                                    c.getId()
+                                                            .substring(
+                                                                    c.getId().lastIndexOf("/") + 1)
+                                                            .trim())));
+        }
 
         communities.addAll(czenodo);
 

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

@@ -67,8 +67,8 @@
             <name-node>${nameNode}</name-node>
             <master>yarn-cluster</master>
             <mode>cluster</mode>
-            <name>bulkTagging</name>
-            <class>eu.dnetlib.dhp.SparkBulkTagJob2</class>
+            <name>bulkTagging-publication</name>
+            <class>eu.dnetlib.dhp.bulktag.SparkBulkTagJob2</class>
             <jar>dhp-bulktag-${projectVersion}.jar</jar>
             <spark-opts>
                 --num-executors=${sparkExecutorNumber}
@@ -96,8 +96,8 @@
             <name-node>${nameNode}</name-node>
             <master>yarn-cluster</master>
             <mode>cluster</mode>
-            <name>bulkTagging</name>
-            <class>eu.dnetlib.dhp.SparkBulkTagJob2</class>
+            <name>bulkTagging-dataset</name>
+            <class>eu.dnetlib.dhp.bulktag.SparkBulkTagJob2</class>
             <jar>dhp-bulktag-${projectVersion}.jar</jar>
             <spark-opts>
                 --num-executors=${sparkExecutorNumber}
@@ -125,8 +125,8 @@
             <name-node>${nameNode}</name-node>
             <master>yarn-cluster</master>
             <mode>cluster</mode>
-            <name>bulkTagging</name>
-            <class>eu.dnetlib.dhp.SparkBulkTagJob2</class>
+            <name>bulkTagging-orp</name>
+            <class>eu.dnetlib.dhp.bulktag.SparkBulkTagJob2</class>
             <jar>dhp-bulktag-${projectVersion}.jar</jar>
             <spark-opts>
                 --num-executors=${sparkExecutorNumber}
@@ -154,8 +154,8 @@
             <name-node>${nameNode}</name-node>
             <master>yarn-cluster</master>
             <mode>cluster</mode>
-            <name>bulkTagging</name>
-            <class>eu.dnetlib.dhp.SparkBulkTagJob2</class>
+            <name>bulkTagging-software</name>
+            <class>eu.dnetlib.dhp.bulktag.SparkBulkTagJob2</class>
             <jar>dhp-bulktag-${projectVersion}.jar</jar>
             <spark-opts>
                 --num-executors=${sparkExecutorNumber}

dhp-workflows/dhp-bulktag/src/test/java/eu/dnetlib/dhp/BulkTagJobTest.java

@@ -3,6 +3,7 @@ package eu.dnetlib.dhp;
 import static eu.dnetlib.dhp.community.TagginConstants.ZENODO_COMMUNITY_INDICATOR;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import eu.dnetlib.dhp.bulktag.SparkBulkTagJob2;
 import eu.dnetlib.dhp.schema.oaf.Dataset;
 import eu.dnetlib.dhp.schema.oaf.OtherResearchProduct;
 import eu.dnetlib.dhp.schema.oaf.Publication;

dhp-workflows/dhp-bulktag/src/test/java/eu/dnetlib/dhp/CommunityConfigurationFactoryTest.java

@@ -141,4 +141,15 @@ public class CommunityConfigurationFactoryTest {
 
         System.out.println(cc.toJson());
     }
+
+    @Test
+    public void temporaneo() throws Exception {
+        String xml =
+                IOUtils.toString(
+                        getClass()
+                                .getResourceAsStream(
+                                        "/eu/dnetlib/dhp/communityconfiguration/tagging_conf.xml"));
+        final CommunityConfiguration cc = CommunityConfigurationFactory.newInstance(xml);
+        System.out.println(cc.toJson());
+    }
 }