new prepareProgramme according to the new file

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/PrepareProgramme.java

@@ -143,7 +143,6 @@ public class PrepareProgramme {
 
 		JavaRDD<CSVProgramme> h2020Programmes = programme
 			.toJavaRDD()
-			.filter(p -> p.getFrameworkProgramme().trim().equalsIgnoreCase("H2020"))
 			.mapToPair(csvProgramme -> new Tuple2<>(csvProgramme.getCode(), csvProgramme))
 			.reduceByKey((a, b) -> {
 				if (!a.getLanguage().equals("en")) {

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/PrepareProjects.java

@@ -32,7 +32,7 @@ public class PrepareProjects {
 
 	private static final Logger log = LoggerFactory.getLogger(PrepareProgramme.class);
 	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-	private static final HashMap<String, CSVProgramme> programmeMap = new HashMap<>();
+	//private static final HashMap<String, CSVProgramme> programmeMap = new HashMap<>();
 
 	public static void main(String[] args) throws Exception {
 

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/SparkAtomicActionJob.java

@@ -120,7 +120,6 @@ public class SparkAtomicActionJob {
 			.map((MapFunction<Tuple2<CSVProject, CSVProgramme>, Project>) c -> {
 
 				CSVProject csvProject = c._1();
-				Optional<CSVProgramme> ocsvProgramme = Optional.ofNullable(c._2());
 
 				return Optional
 					.ofNullable(c._2())
@@ -135,9 +134,9 @@ public class SparkAtomicActionJob {
 						H2020Programme pm = new H2020Programme();
 						H2020Classification h2020classification = new H2020Classification();
 						pm.setCode(csvProject.getProgramme());
-						h2020classification.setClassification(ocsvProgramme.get().getClassification());
+						h2020classification.setClassification(csvProgramme.getClassification());
 						h2020classification.setH2020Programme(pm);
-						setLevelsandProgramme(h2020classification, ocsvProgramme.get().getClassification_short());
+						setLevelsandProgramme(h2020classification, csvProgramme.getClassification_short());
 						// setProgramme(h2020classification, ocsvProgramme.get().getClassification());
 						pp.setH2020classification(Arrays.asList(h2020classification));
 
@@ -145,10 +144,11 @@ public class SparkAtomicActionJob {
 					})
 					.orElse(null);
 
-			}, Encoders.bean(Project.class));
+			}, Encoders.bean(Project.class))
+			.filter(Objects::nonNull);
 
 		aaproject
-			.joinWith(topic, aaproject.col("h2020topiccode").equalTo(topic.col("code")))
+			.joinWith(topic, aaproject.col("h2020topiccode").equalTo(topic.col("code")), "left")
 			.map((MapFunction<Tuple2<Project, EXCELTopic>, Project>) p -> {
 				Optional<EXCELTopic> op = Optional.ofNullable(p._2());
 				Project rp = p._1();

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/utils/CSVProgramme.java

@@ -7,14 +7,7 @@ import java.io.Serializable;
  * The model for the programme csv file
  */
 public class CSVProgramme implements Serializable {
-	private String parentProgramme;
-	private String frameworkProgramme;
-	private String startDate;
-	private String endDate;
-	private String objective;
-	private String subjects;
-	private String legalBasis;
-	private String call;
+
 	private String rcn;
 	private String code;
 
@@ -80,67 +73,5 @@ public class CSVProgramme implements Serializable {
 		this.language = language;
 	}
 
-	public String getParentProgramme() {
-		return parentProgramme;
-	}
-
-	public void setParentProgramme(String parentProgramme) {
-		this.parentProgramme = parentProgramme;
-	}
-
-	public String getFrameworkProgramme() {
-		return frameworkProgramme;
-	}
-
-	public void setFrameworkProgramme(String frameworkProgramme) {
-		this.frameworkProgramme = frameworkProgramme;
-	}
-
-	public String getStartDate() {
-		return startDate;
-	}
-
-	public void setStartDate(String startDate) {
-		this.startDate = startDate;
-	}
-
-	public String getEndDate() {
-		return endDate;
-	}
-
-	public void setEndDate(String endDate) {
-		this.endDate = endDate;
-	}
-
-	public String getObjective() {
-		return objective;
-	}
-
-	public void setObjective(String objective) {
-		this.objective = objective;
-	}
-
-	public String getSubjects() {
-		return subjects;
-	}
-
-	public void setSubjects(String subjects) {
-		this.subjects = subjects;
-	}
-
-	public String getLegalBasis() {
-		return legalBasis;
-	}
-
-	public void setLegalBasis(String legalBasis) {
-		this.legalBasis = legalBasis;
-	}
-
-	public String getCall() {
-		return call;
-	}
-
-	public void setCall(String call) {
-		this.call = call;
-	}
+//
 }

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/actionmanager/project/oozie_app/workflow.xml

@@ -1,4 +1,4 @@
-<workflow-app name="H2020Programme" xmlns="uri:oozie:workflow:0.5">
+<workflow-app name="H2020Classification" xmlns="uri:oozie:workflow:0.5">
     <parameters>
         <property>
             <name>projectFileURL</name>
@@ -35,10 +35,17 @@
             <delete path='${workingDir}'/>
             <mkdir path='${workingDir}'/>
         </fs>
-        <ok to="get_project_file"/>
+        <ok to="fork_get_info"/>
         <error to="Kill"/>
     </action>
 
+
+    <fork name="fork_get_info">
+        <path start="get_project_file"/>
+        <path start="get_programme_file"/>
+        <path start="get_topic_file"/>
+        <path start="read_projects"/>
+    </fork>
     <action name="get_project_file">
         <java>
             <main-class>eu.dnetlib.dhp.actionmanager.project.utils.ReadCSV</main-class>
@@ -47,7 +54,7 @@
             <arg>--hdfsPath</arg><arg>${workingDir}/projects</arg>
             <arg>--classForName</arg><arg>eu.dnetlib.dhp.actionmanager.project.utils.CSVProject</arg>
         </java>
-        <ok to="get_programme_file"/>
+        <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
@@ -59,7 +66,7 @@
             <arg>--hdfsPath</arg><arg>${workingDir}/programme</arg>
             <arg>--classForName</arg><arg>eu.dnetlib.dhp.actionmanager.project.utils.CSVProgramme</arg>
         </java>
-        <ok to="get_topic_file"/>
+        <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
@@ -72,7 +79,7 @@
             <arg>--sheetName</arg><arg>${sheetName}</arg>
             <arg>--classForName</arg><arg>eu.dnetlib.dhp.actionmanager.project.utils.EXCELTopic</arg>
         </java>
-        <ok to="read_projects"/>
+        <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
@@ -85,10 +92,12 @@
             <arg>--postgresUser</arg><arg>${postgresUser}</arg>
             <arg>--postgresPassword</arg><arg>${postgresPassword}</arg>
         </java>
-        <ok to="prepare_programme"/>
+        <ok to="wait"/>
         <error to="Kill"/>
     </action>
 
+    <join name="wait" to="prepare_programme"/>
+
     <action name="prepare_programme">
         <spark xmlns="uri:oozie:spark-action:0.2">
             <master>yarn</master>