fix dataset creation for downloaded works

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/orcid/OrcidData.java

@@ -3,6 +3,8 @@ package eu.dnetlib.dhp.schema.orcid;
 
 public class OrcidData {
 	protected String base64CompressData;
+	protected String statusCode;
+	protected String downloadDate;
 
 	public String getBase64CompressData() {
 		return base64CompressData;
@@ -11,4 +13,20 @@ public class OrcidData {
 	public void setBase64CompressData(String base64CompressData) {
 		this.base64CompressData = base64CompressData;
 	}
+
+	public String getStatusCode() {
+		return statusCode;
+	}
+
+	public void setStatusCode(String statusCode) {
+		this.statusCode = statusCode;
+	}
+
+	public String getDownloadDate() {
+		return downloadDate;
+	}
+
+	public void setDownloadDate(String downloadDate) {
+		this.downloadDate = downloadDate;
+	}
 }

dhp-workflows/dhp-doiboost/src/main/java/eu/dnetlib/doiboost/orcid/SparkUpdateOrcidDatasets.java

@@ -4,30 +4,47 @@ package eu.dnetlib.doiboost.orcid;
 import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
 
 import java.io.IOException;
-import java.util.Optional;
+import java.util.*;
 
 import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.compress.GzipCodec;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.PairFunction;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.util.LongAccumulator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.schema.orcid.AuthorSummary;
 import eu.dnetlib.dhp.schema.orcid.Work;
 import eu.dnetlib.dhp.schema.orcid.WorkDetail;
+import eu.dnetlib.dhp.utils.DHPUtils;
 import eu.dnetlib.doiboost.orcid.xml.XMLRecordParser;
 import eu.dnetlib.doiboost.orcidnodoi.json.JsonWriter;
 import eu.dnetlib.doiboost.orcidnodoi.xml.XMLRecordParserNoDoi;
+import scala.Tuple2;
 
 public class SparkUpdateOrcidDatasets {
 
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
 	public static void main(String[] args) throws IOException, Exception {
 		Logger logger = LoggerFactory.getLogger(SparkUpdateOrcidDatasets.class);
-		logger.info("[ SparkUpdateOrcidDatasets STARTED]");
 
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
 			IOUtils
@@ -40,11 +57,8 @@ public class SparkUpdateOrcidDatasets {
 			.ofNullable(parser.get("isSparkSessionManaged"))
 			.map(Boolean::valueOf)
 			.orElse(Boolean.TRUE);
-		logger.info("isSparkSessionManaged: {}", isSparkSessionManaged);
 		final String workingPath = parser.get("workingPath");
-		logger.info("workingPath: ", workingPath);
 //		final String outputPath = parser.get("outputPath");
-//		logger.info("outputPath: ", outputPath);
 
 		SparkConf conf = new SparkConf();
 		runWithSparkSession(
@@ -53,35 +67,247 @@ public class SparkUpdateOrcidDatasets {
 			spark -> {
 				JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
 
-				JavaPairRDD<Text, Text> xmlSummariesRDD = sc
+				LongAccumulator errorCodeAuthorsFoundAcc = spark
-					.sequenceFile(workingPath.concat("xml/authors/xml_authors.seq"), Text.class, Text.class);
+					.sparkContext()
-				xmlSummariesRDD
+					.longAccumulator("error_code_authors_found");
-					.map(seq -> {
+				LongAccumulator errorLoadingAuthorsJsonFoundAcc = spark
-						AuthorSummary authorSummary = XMLRecordParser
+					.sparkContext()
-							.VTDParseAuthorSummary(seq._2().toString().getBytes());
+					.longAccumulator("error_loading_authors_json_found");
-						authorSummary
+				LongAccumulator errorLoadingAuthorsXMLFoundAcc = spark
-							.setBase64CompressData(ArgumentApplicationParser.compressArgument(seq._2().toString()));
+					.sparkContext()
-						return authorSummary;
+					.longAccumulator("error_loading_authors_xml_found");
-					})
+				LongAccumulator errorParsingAuthorsXMLFoundAcc = spark
-					.filter(authorSummary -> authorSummary != null)
+					.sparkContext()
-					.map(authorSummary -> JsonWriter.create(authorSummary))
+					.longAccumulator("error_parsing_authors_xml_found");
-					.saveAsTextFile(workingPath.concat("orcid_dataset/authors"), GzipCodec.class);
 
-				JavaPairRDD<Text, Text> xmlWorksRDD = sc
+				LongAccumulator updatedWorksFoundAcc = spark
-					.sequenceFile(workingPath.concat("xml/works/*"), Text.class, Text.class);
+					.sparkContext()
+					.longAccumulator("updated_works_found");
+				LongAccumulator errorCodeWorksFoundAcc = spark
+					.sparkContext()
+					.longAccumulator("error_code_works_found");
+				LongAccumulator errorLoadingWorksJsonFoundAcc = spark
+					.sparkContext()
+					.longAccumulator("error_loading_works_json_found");
+				LongAccumulator errorLoadingWorksXMLFoundAcc = spark
+					.sparkContext()
+					.longAccumulator("error_loading_works_xml_found");
+				LongAccumulator errorParsingWorksXMLFoundAcc = spark
+					.sparkContext()
+					.longAccumulator("error_parsing_works_xml_found");
 
-				xmlWorksRDD
+//				JavaPairRDD<Text, Text> xmlSummariesRDD = sc
-					.map(seq -> {
+//					.sequenceFile(workingPath.concat("xml/authors/xml_authors.seq"), Text.class, Text.class);
-						WorkDetail workDetail = XMLRecordParserNoDoi.VTDParseWorkData(seq._2().toString().getBytes());
+//				xmlSummariesRDD
-						Work work = new Work();
+//					.map(seq -> {
-						work.setWorkDetail(workDetail);
+//						AuthorSummary authorSummary = XMLRecordParser
-						work.setBase64CompressData(ArgumentApplicationParser.compressArgument(seq._2().toString()));
+//							.VTDParseAuthorSummary(seq._2().toString().getBytes());
-						return work;
+//						authorSummary
-					})
+//							.setBase64CompressData(ArgumentApplicationParser.compressArgument(seq._2().toString()));
-					.filter(work -> work != null)
+//						return authorSummary;
+//					})
+//					.filter(authorSummary -> authorSummary != null)
+//					.map(authorSummary -> JsonWriter.create(authorSummary))
+//					.saveAsTextFile(workingPath.concat("orcid_dataset/authors"), GzipCodec.class);
+//
+//				JavaPairRDD<Text, Text> xmlWorksRDD = sc
+//					.sequenceFile(workingPath.concat("xml/works/*"), Text.class, Text.class);
+//
+//				xmlWorksRDD
+//					.map(seq -> {
+//						WorkDetail workDetail = XMLRecordParserNoDoi.VTDParseWorkData(seq._2().toString().getBytes());
+//						Work work = new Work();
+//						work.setWorkDetail(workDetail);
+//						work.setBase64CompressData(ArgumentApplicationParser.compressArgument(seq._2().toString()));
+//						return work;
+//					})
+//					.filter(work -> work != null)
+//					.map(work -> JsonWriter.create(work))
+//					.saveAsTextFile(workingPath.concat("orcid_dataset/works"), GzipCodec.class);
+
+//				Function<Tuple2<Text, Text>, AuthorSummary> retrieveAuthorSummaryFunction = data -> {
+//					AuthorSummary authorSummary = new AuthorSummary();
+//					String orcidId = data._1().toString();
+//					String jsonData = data._2().toString();
+//					JsonElement jElement = new JsonParser().parse(jsonData);
+//					String statusCode = getJsonValue(jElement, "statusCode");
+//					String downloadDate = getJsonValue(jElement, "lastModifiedDate");
+//					if (statusCode.equals("200")) {
+//						String compressedData = getJsonValue(jElement, "compressedData");
+//						if (StringUtils.isEmpty(compressedData)) {
+//							errorLoadingAuthorsJsonFoundAcc.add(1);
+//						} else {
+//							String xmlAuthor = ArgumentApplicationParser.decompressValue(compressedData);
+//							if (StringUtils.isEmpty(xmlAuthor)) {
+//								errorLoadingAuthorsXMLFoundAcc.add(1);
+//							} else {
+//								try {
+//									authorSummary = XMLRecordParser
+//										.VTDParseAuthorSummary(xmlAuthor.getBytes());
+//									authorSummary.setStatusCode(statusCode);
+//									authorSummary.setDownloadDate(downloadDate);
+//									authorSummary.setBase64CompressData(compressedData);
+//									return authorSummary;
+//								} catch (Exception e) {
+//									logger.error("parsing xml " + orcidId + " [" + jsonData + "]", e);
+//									errorParsingAuthorsXMLFoundAcc.add(1);
+//								}
+//							}
+//						}
+//					} else {
+//						authorSummary.setStatusCode(statusCode);
+//						authorSummary.setDownloadDate(downloadDate);
+//						errorCodeAuthorsFoundAcc.add(1);
+//					}
+//					return authorSummary;
+//				};
+//
+//				Dataset<AuthorSummary> downloadedAuthorSummaryDS = spark
+//					.createDataset(
+//						sc
+//							.sequenceFile(workingPath + "downloads/updated_authors/*", Text.class, Text.class)
+//							.map(retrieveAuthorSummaryFunction)
+//							.rdd(),
+//						Encoders.bean(AuthorSummary.class));
+//				Dataset<AuthorSummary> currentAuthorSummaryDS = spark
+//					.createDataset(
+//						sc
+//							.textFile(workingPath.concat("orcid_dataset/authors/*"))
+//							.map(item -> OBJECT_MAPPER.readValue(item, AuthorSummary.class))
+//							.rdd(),
+//						Encoders.bean(AuthorSummary.class));
+//				currentAuthorSummaryDS
+//					.joinWith(
+//						downloadedAuthorSummaryDS,
+//						currentAuthorSummaryDS
+//							.col("authorData.oid")
+//							.equalTo(downloadedAuthorSummaryDS.col("authorData.oid")),
+//						"full_outer")
+//					.map(value -> {
+//						Optional<AuthorSummary> opCurrent = Optional.ofNullable(value._1());
+//						Optional<AuthorSummary> opDownloaded = Optional.ofNullable(value._2());
+//						if (!opCurrent.isPresent()) {
+//							return opDownloaded.get();
+//						}
+//						if (!opDownloaded.isPresent()) {
+//							return opCurrent.get();
+//						}
+//						if (opCurrent.isPresent() && opDownloaded.isPresent()) {
+//							return opDownloaded.get();
+//						}
+//						return null;
+//					},
+//						Encoders.bean(AuthorSummary.class))
+//					.filter(Objects::nonNull)
+//					.toJavaRDD()
+//					.map(authorSummary -> JsonWriter.create(authorSummary))
+//					.saveAsTextFile(workingPath.concat("orcid_dataset/new_authors"), GzipCodec.class);
+//
+//				logger.info("errorCodeFoundAcc: " + errorCodeAuthorsFoundAcc.value().toString());
+//				logger.info("errorLoadingJsonFoundAcc: " + errorLoadingAuthorsJsonFoundAcc.value().toString());
+//				logger.info("errorLoadingXMLFoundAcc: " + errorLoadingAuthorsXMLFoundAcc.value().toString());
+//				logger.info("errorParsingXMLFoundAcc: " + errorParsingAuthorsXMLFoundAcc.value().toString());
+
+				Function<String, Work> retrieveWorkFunction = jsonData -> {
+					Work work = new Work();
+					JsonElement jElement = new JsonParser().parse(jsonData);
+					String statusCode = getJsonValue(jElement, "statusCode");
+					work.setStatusCode(statusCode);
+					String downloadDate = getJsonValue(jElement, "lastModifiedDate");
+					work.setDownloadDate(downloadDate);
+					if (statusCode.equals("200")) {
+						String compressedData = getJsonValue(jElement, "compressedData");
+						if (StringUtils.isEmpty(compressedData)) {
+							errorLoadingWorksJsonFoundAcc.add(1);
+						} else {
+							String xmlWork = ArgumentApplicationParser.decompressValue(compressedData);
+							if (StringUtils.isEmpty(xmlWork)) {
+								errorLoadingWorksXMLFoundAcc.add(1);
+							} else {
+								try {
+									WorkDetail workDetail = XMLRecordParserNoDoi
+										.VTDParseWorkData(xmlWork.getBytes());
+									work.setWorkDetail(workDetail);
+									work.setBase64CompressData(compressedData);
+									updatedWorksFoundAcc.add(1);
+									return work;
+								} catch (Exception e) {
+									logger.error("parsing xml [" + jsonData + "]", e);
+									errorParsingWorksXMLFoundAcc.add(1);
+								}
+							}
+						}
+					} else {
+						errorCodeWorksFoundAcc.add(1);
+					}
+					return work;
+				};
+
+				Dataset<Work> downloadedWorksDS = spark
+					.createDataset(
+						sc
+							.textFile(workingPath + "downloads/updated_works/*")
+							.map(s -> {
+								return s.substring(21, s.length() - 1);
+							})
+							.map(retrieveWorkFunction)
+							.rdd(),
+						Encoders.bean(Work.class));
+				Dataset<Work> currentWorksDS = spark
+					.createDataset(
+						sc
+							.textFile(workingPath.concat("orcid_dataset/works/*"))
+							.map(item -> OBJECT_MAPPER.readValue(item, Work.class))
+							.rdd(),
+						Encoders.bean(Work.class));
+				currentWorksDS
+					.joinWith(
+						downloadedWorksDS,
+						currentWorksDS
+							.col("workDetail.id")
+							.equalTo(downloadedWorksDS.col("workDetail.id"))
+							.and(
+								currentWorksDS
+									.col("workDetail.oid")
+									.equalTo(downloadedWorksDS.col("workDetail.oid"))),
+						"full_outer")
+					.map(value -> {
+						Optional<Work> opCurrent = Optional.ofNullable(value._1());
+						Optional<Work> opDownloaded = Optional.ofNullable(value._2());
+						if (!opCurrent.isPresent()) {
+							return opDownloaded.get();
+						}
+						if (!opDownloaded.isPresent()) {
+							return opCurrent.get();
+						}
+						if (opCurrent.isPresent() && opDownloaded.isPresent()) {
+							return opDownloaded.get();
+						}
+						return null;
+					},
+						Encoders.bean(Work.class))
+					.filter(Objects::nonNull)
+					.toJavaRDD()
 					.map(work -> JsonWriter.create(work))
-					.saveAsTextFile(workingPath.concat("orcid_dataset/works"), GzipCodec.class);
+					.saveAsTextFile(workingPath.concat("orcid_dataset/new_works"), GzipCodec.class);
-			});
 
+				logger.info("updatedWorksFoundAcc: " + updatedWorksFoundAcc.value().toString());
+				logger.info("errorCodeWorksFoundAcc: " + errorCodeWorksFoundAcc.value().toString());
+				logger.info("errorLoadingJsonWorksFoundAcc: " + errorLoadingWorksJsonFoundAcc.value().toString());
+				logger.info("errorLoadingXMLWorksFoundAcc: " + errorLoadingWorksXMLFoundAcc.value().toString());
+				logger.info("errorParsingXMLWorksFoundAcc: " + errorParsingWorksXMLFoundAcc.value().toString());
+
+			});
+	}
+
+	private static String getJsonValue(JsonElement jElement, String property) {
+		if (jElement.getAsJsonObject().has(property)) {
+			JsonElement name = null;
+			name = jElement.getAsJsonObject().get(property);
+			if (name != null && !name.isJsonNull()) {
+				return name.getAsString();
+			}
+		}
+		return "";
 	}
 }

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

@@ -2,7 +2,7 @@
     <parameters>
         <property>
             <name>spark2MaxExecutors</name>
-            <value>40</value>
+            <value>50</value>
         </property>
         <property>
             <name>sparkDriverMemory</name>

dhp-workflows/dhp-doiboost/src/test/java/eu/dnetlib/doiboost/orcid/OrcidClientTest.java

@@ -372,12 +372,17 @@ public class OrcidClientTest {
 	}
 
 	@Test
-	public void testReadDownloadedWork() throws Exception {
+	public void testUpdatedRecord() throws Exception {
 		final String base64CompressedRecord = IOUtils
-			.toString(getClass().getResourceAsStream("0000-0002-6664-7451_work.compressed.base64"));
+			.toString(getClass().getResourceAsStream("0000-0003-3028-6161.compressed.base64"));
-		final String work = ArgumentApplicationParser.decompressValue(base64CompressedRecord);
+		final String record = ArgumentApplicationParser.decompressValue(base64CompressedRecord);
-		logToFile("\n\ndownloaded \n\n" + work);
+		logToFile("\n\nrecord updated \n\n" + record);
-//		final String downloadedRecord = testDownloadRecord("0000-0003-3028-6161", REQUEST_TYPE_RECORD);
+	}
-//		assertTrue(recordFromSeqFile.equals(downloadedRecord));
+
+	@Test
+	public void testUpdatedWork() throws Exception {
+		final String base64CompressedWork = "H4sIAAAAAAAAAM1XS2/jNhC+51cQOuxJsiXZSR03Vmq0G6Bo013E6R56oyXaZiOJWpKy4y783zvUg5Ksh5uiCJogisX5Zjj85sHx3f1rFKI94YKyeGE4I9tAJPZZQOPtwvj9+cGaGUhIHAc4ZDFZGEcijHvv6u7A+MtcPVCSSgsUQObYzuzaccBEguVuYYxt+LHgbwKP6a11M3WnY6UzrpB7KuiahlQeF0aSrkPqGwhcisWcxpLwGIcLYydlMh+PD4fDiHGfBvDcjmMxLhGlBglSH8vsIH0qGlLqBFRIGvvDWjWQ1iMJJ2CKBANqGlNqMbkj3IpxRPq1KkypFZFoDRHa0aRfq8JoNjhnfIAJJS6xPouiIQJyeYmGQzE+cO5cXqITcItBlKyASExD0a93jiwtvJDjYXDDAqBPHoH2wMmVWGNf8xyyaEBiSTeUDHHWBpd2Nmmc10yfbgHQrHCyIRxKjQwRUoFKPRwEnIgBnQJQVdGeQgJaCRN0OMnPkaUFVbD9WkpaIndQJowf+8EFoIpTErJjBFQOBavElFpfUxwC9ZcqvQErdQXhe+oPFF8BaObupYzVsYEOARzSoZBWmKqaBMHcV0Wf8oG0beIqD+Gdkz0lhyE3NajUW6fhQFSV9Nw/MCBYyofYa0EN7wrBz13eP+Y+J6obWgE8Pdd2JpYD94P77Ezmjj13b0bu5PqPu3EXumEnxEJaEVxSUIHammsra+53z44zt2/m1/bItaeVtQ6dhs3c4XytvW75IYUchMKvEHVUyqmnWBFAS0VJrqSvQde6vp251ux2NtFuKcVOi+oK9YY0M0Cn6o4J6WkvtEK2XJ1vfPGAZxSoK8lb+SxJBbLQx1CohOLndjJUywQWUFmqEi3G6Zaqf/7buOyYJd5IYpfmf0XipfP18pDR9cQCeEuJQI/Lx36bFbVnpBeL2UwmqQw7ApAvf4GeGGQdEbENgolui/wdpjHaYCmPCIPPAmGBIsxfoLUhyRCB0SeCakEBJRKBtfJ+UBbI15TG4PaGBAhWthx8DmFYtHZQujv1CWbLLdzmmUKmHEOWCe1/zdu78bn/+YH+hCOqOzcXfFwuP6OVT/P710crwqGXFrpNaM2GT3MXarw01i15TIi3pmtJXgtbTVGf3h6HKfF+wBAnPyTfdCChudlm5gZaoG//F9pPZsGQcqqbyZN5hBau5OoIJ3PPwjTKDuG4s5MZp2rMzF5PZoK34IT6PIFOPrk+mTiVO5aJH2C+JJRjE/06eoRfpJxa4VgyYaLlaJUv/EhCfATMU/76gEOfmehL/qbJNNHjaFna+CQYB8wvo9PpPFJ5MOrJ1Ix7USBZqBl7KRNOx1d3jex7SG6zuijqCMWRusBsncjZSrM2u82UJmqzpGhvUJN2t6caIM9QQgO9c0t40UROnWsJd2Rbs+nsxpna9u30ttNkjechmzHjEST+X5CkkuNY0GzQkzyFseAf7lSZuLwdh1xSXKvvQJ4g4abTYgPV7uMt3rskohlJmMa82kQkshtyBEIYqQ+YB8X3oRHg7iFKi/bZP+Ao+T6BJhIT/vNPi8ffZs+flk+r2v0WNroZiyWn6xRmadHqTJXsjLJczElAZX6TnJdoWTM1SI2gfutv3rjeBt5t06rVvNuWup29246tlvluO+u2/G92bK9DXheL6uFd/Q3EaRDZqBIAAA==";
+		final String work = ArgumentApplicationParser.decompressValue(base64CompressedWork);
+		logToFile("\n\nwork updated \n\n" + work);
 	}
 }