Merge pull request '[UsageCount] addition of usagecount for Projects and datasources' (#296) from master_datasource_project_usagecounts into master

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

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/usagestats/SparkAtomicActionUsageJob.java

@@ -14,7 +14,6 @@ import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.SequenceFileOutputFormat;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.MapFunction;
-import org.apache.spark.api.java.function.MapGroupsFunction;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.SaveMode;
@@ -28,9 +27,7 @@ import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import eu.dnetlib.dhp.common.HdfsSupport;
 import eu.dnetlib.dhp.schema.action.AtomicAction;
 import eu.dnetlib.dhp.schema.common.ModelConstants;
-import eu.dnetlib.dhp.schema.oaf.DataInfo;
+import eu.dnetlib.dhp.schema.oaf.*;
-import eu.dnetlib.dhp.schema.oaf.Measure;
-import eu.dnetlib.dhp.schema.oaf.Result;
 import eu.dnetlib.dhp.schema.oaf.utils.OafMapperUtils;
 import scala.Tuple2;
 
@@ -76,16 +73,22 @@ public class SparkAtomicActionUsageJob implements Serializable {
 			isSparkSessionManaged,
 			spark -> {
 				removeOutputDir(spark, outputPath);
-				prepareResults(dbname, spark, workingPath);
+				prepareData(dbname, spark, workingPath + "/usageDb", "usage_stats", "result_id");
+				prepareData(dbname, spark, workingPath + "/projectDb", "project_stats", "id");
+				prepareData(dbname, spark, workingPath + "/datasourceDb", "datasource_stats", "repositor_id");
 				writeActionSet(spark, workingPath, outputPath);
 			});
 	}
 
-	public static void prepareResults(String db, SparkSession spark, String workingPath) {
+	private static void prepareData(String dbname, SparkSession spark, String workingPath, String tableName,
+		String attribute_name) {
 		spark
 			.sql(
-				"Select result_id, downloads, views " +
+				String
-					"from " + db + ".usage_stats")
+					.format(
+						"select %s as id, sum(downloads) as downloads, sum(views) as views " +
+							"from %s.%s group by %s",
+						attribute_name, dbname, tableName, attribute_name))
 			.as(Encoders.bean(UsageStatsModel.class))
 			.write()
 			.mode(SaveMode.Overwrite)
@@ -94,23 +97,17 @@ public class SparkAtomicActionUsageJob implements Serializable {
 	}
 
 	public static void writeActionSet(SparkSession spark, String inputPath, String outputPath) {
-		readPath(spark, inputPath, UsageStatsModel.class)
+		getFinalIndicatorsResult(spark, inputPath + "/usageDb")
-			.groupByKey((MapFunction<UsageStatsModel, String>) us -> us.getResult_id(), Encoders.STRING())
-			.mapGroups((MapGroupsFunction<String, UsageStatsModel, Result>) (k, it) -> {
-				UsageStatsModel first = it.next();
-				it.forEachRemaining(us -> {
-					first.setDownloads(first.getDownloads() + us.getDownloads());
-					first.setViews(first.getViews() + us.getViews());
-				});
-
-				Result res = new Result();
-				res.setId("50|" + k);
-
-				res.setMeasures(getMeasure(first.getDownloads(), first.getViews()));
-				return res;
-			}, Encoders.bean(Result.class))
 			.toJavaRDD()
 			.map(p -> new AtomicAction(p.getClass(), p))
+			.union(
+				getFinalIndicatorsProject(spark, inputPath + "/projectDb")
+					.toJavaRDD()
+					.map(p -> new AtomicAction(p.getClass(), p)))
+			.union(
+				getFinalIndicatorsDatasource(spark, inputPath + "/datasourceDb")
+					.toJavaRDD()
+					.map(p -> new AtomicAction(p.getClass(), p)))
 			.mapToPair(
 				aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
 					new Text(OBJECT_MAPPER.writeValueAsString(aa))))
@@ -118,6 +115,39 @@ public class SparkAtomicActionUsageJob implements Serializable {
 
 	}
 
+	private static Dataset<Result> getFinalIndicatorsResult(SparkSession spark, String inputPath) {
+
+		return readPath(spark, inputPath, UsageStatsModel.class)
+			.map((MapFunction<UsageStatsModel, Result>) usm -> {
+				Result r = new Result();
+				r.setId("50|" + usm.getId());
+				r.setMeasures(getMeasure(usm.getDownloads(), usm.getViews()));
+				return r;
+			}, Encoders.bean(Result.class));
+	}
+
+	private static Dataset<Project> getFinalIndicatorsProject(SparkSession spark, String inputPath) {
+
+		return readPath(spark, inputPath, UsageStatsModel.class)
+			.map((MapFunction<UsageStatsModel, Project>) usm -> {
+				Project p = new Project();
+				p.setId("40|" + usm.getId());
+				p.setMeasures(getMeasure(usm.getDownloads(), usm.getViews()));
+				return p;
+			}, Encoders.bean(Project.class));
+	}
+
+	private static Dataset<Datasource> getFinalIndicatorsDatasource(SparkSession spark, String inputPath) {
+
+		return readPath(spark, inputPath, UsageStatsModel.class)
+			.map((MapFunction<UsageStatsModel, Datasource>) usm -> {
+				Datasource d = new Datasource();
+				d.setId("10|" + usm.getId());
+				d.setMeasures(getMeasure(usm.getDownloads(), usm.getViews()));
+				return d;
+			}, Encoders.bean(Datasource.class));
+	}
+
 	private static List<Measure> getMeasure(Long downloads, Long views) {
 		DataInfo dataInfo = OafMapperUtils
 			.dataInfo(

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/usagestats/UsageStatsModel.java

@@ -4,16 +4,16 @@ package eu.dnetlib.dhp.actionmanager.usagestats;
 import java.io.Serializable;
 
 public class UsageStatsModel implements Serializable {
-	private String result_id;
+	private String id;
 	private Long downloads;
 	private Long views;
 
-	public String getResult_id() {
+	public String getId() {
-		return result_id;
+		return id;
 	}
 
-	public void setResult_id(String result_id) {
+	public void setId(String id) {
-		this.result_id = result_id;
+		this.id = id;
 	}
 
 	public Long getDownloads() {

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

@@ -89,7 +89,7 @@
             <arg>--hive_metastore_uris</arg><arg>${hiveMetastoreUris}</arg>
             <arg>--outputPath</arg><arg>${outputPath}</arg>
             <arg>--usagestatsdb</arg><arg>${usagestatsdb}</arg>
-            <arg>--workingPath</arg><arg>${workingDir}/usageDb</arg>
+            <arg>--workingPath</arg><arg>${workingDir}</arg>
         </spark>
         <ok to="End"/>
         <error to="Kill"/>

dhp-workflows/dhp-aggregation/src/test/java/eu/dnetlib/dhp/actionmanager/usagestats/SparkAtomicActionCountJobTest.java

@@ -24,6 +24,7 @@ import org.slf4j.LoggerFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.OafEntity;
 import eu.dnetlib.dhp.schema.oaf.Result;
 
 public class SparkAtomicActionCountJobTest {
@@ -68,24 +69,26 @@ public class SparkAtomicActionCountJobTest {
 	@Test
 	void testMatch() {
 		String usageScoresPath = getClass()
-			.getResource("/eu/dnetlib/dhp/actionmanager/usagestats/usagestatsdb")
+			.getResource("/eu/dnetlib/dhp/actionmanager/usagestats")
 			.getPath();
 
 		SparkAtomicActionUsageJob.writeActionSet(spark, usageScoresPath, workingDir.toString() + "/actionSet");
 
 		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
 
-		JavaRDD<Result> tmp = sc
+		JavaRDD<AtomicAction> tmp = sc
 			.sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
-			.map(usm -> OBJECT_MAPPER.readValue(usm._2.getBytes(), AtomicAction.class))
+			.map(usm -> OBJECT_MAPPER.readValue(usm._2.getBytes(), AtomicAction.class));
-			.map(aa -> (Result) aa.getPayload());
+		// .map(aa -> (Result) aa.getPayload());
 
-		Assertions.assertEquals(9, tmp.count());
+		Assertions.assertEquals(9, tmp.filter(aa -> ((OafEntity) aa.getPayload()).getId().startsWith("50|")).count());
+		Assertions.assertEquals(9, tmp.filter(aa -> ((OafEntity) aa.getPayload()).getId().startsWith("10|")).count());
+		Assertions.assertEquals(9, tmp.filter(aa -> ((OafEntity) aa.getPayload()).getId().startsWith("40|")).count());
 
-		tmp.foreach(r -> Assertions.assertEquals(2, r.getMeasures().size()));
+		tmp.foreach(r -> Assertions.assertEquals(2, ((OafEntity) r.getPayload()).getMeasures().size()));
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
@@ -95,14 +98,14 @@ public class SparkAtomicActionCountJobTest {
 							.forEach(u -> Assertions.assertFalse(u.getDataInfo().getDeletedbyinference()))));
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
 						m -> m.getUnit().stream().forEach(u -> Assertions.assertTrue(u.getDataInfo().getInferred()))));
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
@@ -113,7 +116,7 @@ public class SparkAtomicActionCountJobTest {
 
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
@@ -127,7 +130,7 @@ public class SparkAtomicActionCountJobTest {
 										u.getDataInfo().getProvenanceaction().getClassid()))));
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
@@ -142,7 +145,7 @@ public class SparkAtomicActionCountJobTest {
 
 		tmp
 			.foreach(
-				r -> r
+				r -> ((OafEntity) r.getPayload())
 					.getMeasures()
 					.stream()
 					.forEach(
@@ -157,12 +160,19 @@ public class SparkAtomicActionCountJobTest {
 
 		Assertions
 			.assertEquals(
-				1, tmp.filter(r -> r.getId().equals("50|dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6")).count());
+				1,
+				tmp
+					.filter(
+						r -> ((OafEntity) r.getPayload())
+							.getId()
+							.equals("50|dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6"))
+					.count());
 
 		Assertions
 			.assertEquals(
 				"0",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6"))
 					.collect()
 					.get(0)
@@ -178,6 +188,7 @@ public class SparkAtomicActionCountJobTest {
 			.assertEquals(
 				"5",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6"))
 					.collect()
 					.get(0)
@@ -194,6 +205,7 @@ public class SparkAtomicActionCountJobTest {
 			.assertEquals(
 				"0",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|doi_________::17eda2ff77407538fbe5d3d719b9d1c0"))
 					.collect()
 					.get(0)
@@ -209,6 +221,7 @@ public class SparkAtomicActionCountJobTest {
 			.assertEquals(
 				"1",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|doi_________::17eda2ff77407538fbe5d3d719b9d1c0"))
 					.collect()
 					.get(0)
@@ -225,6 +238,7 @@ public class SparkAtomicActionCountJobTest {
 			.assertEquals(
 				"2",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|doi_________::3085e4c6e051378ca6157fe7f0430c1f"))
 					.collect()
 					.get(0)
@@ -240,6 +254,7 @@ public class SparkAtomicActionCountJobTest {
 			.assertEquals(
 				"6",
 				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
 					.filter(r -> r.getId().equals("50|doi_________::3085e4c6e051378ca6157fe7f0430c1f"))
 					.collect()
 					.get(0)
@@ -251,6 +266,204 @@ public class SparkAtomicActionCountJobTest {
 					.getUnit()
 					.get(0)
 					.getValue());
+
+		Assertions
+			.assertEquals(
+				"0",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f1__________::53575dc69e9ace947e02d47ecd54a7a6"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"5",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f1__________::53575dc69e9ace947e02d47ecd54a7a6"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+
+		Assertions
+			.assertEquals(
+				"0",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f11_________::17eda2ff77407538fbe5d3d719b9d1c0"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"1",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f11_________::17eda2ff77407538fbe5d3d719b9d1c0"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+
+		Assertions
+			.assertEquals(
+				"2",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f12_________::3085e4c6e051378ca6157fe7f0430c1f"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"6",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("40|f12_________::3085e4c6e051378ca6157fe7f0430c1f"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+
+		Assertions
+			.assertEquals(
+				"0",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d1__________::53575dc69e9ace947e02d47ecd54a7a6"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"5",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d1__________::53575dc69e9ace947e02d47ecd54a7a6"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+
+		Assertions
+			.assertEquals(
+				"0",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d11_________::17eda2ff77407538fbe5d3d719b9d1c0"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"1",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d11_________::17eda2ff77407538fbe5d3d719b9d1c0"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+
+		Assertions
+			.assertEquals(
+				"2",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d12_________::3085e4c6e051378ca6157fe7f0430c1f"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("downloads"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
+		Assertions
+			.assertEquals(
+				"6",
+				tmp
+					.map(r -> ((OafEntity) r.getPayload()))
+					.filter(r -> r.getId().equals("10|d12_________::3085e4c6e051378ca6157fe7f0430c1f"))
+					.collect()
+					.get(0)
+					.getMeasures()
+					.stream()
+					.filter(m -> m.getId().equals("views"))
+					.collect(Collectors.toList())
+					.get(0)
+					.getUnit()
+					.get(0)
+					.getValue());
 	}
 
 }

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/datasourceDb

@@ -0,0 +1,9 @@
+{"id":"d1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":5}
+{"id":"d11_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"d11_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"d11_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"d12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":6}
+{"id":"d12_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"d12_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"d13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":10}
+{"id":"d13_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/datasourceDb_old

@@ -0,0 +1,12 @@
+{"id":"d1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":4}
+{"id":"d1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":1}
+{"id":"d11_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"d11_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"d11_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"d12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":3}
+{"id":"d12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":0,"views":3}
+{"id":"d12_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"d12_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"d13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":8}
+{"id":"d13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":0,"views":2}
+{"id":"d13_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/projectDb

@@ -0,0 +1,9 @@
+{"id":"f1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":5}
+{"id":"f11_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"f11_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"f11_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"f12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":6}
+{"id":"f12_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"f12_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"f13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":10}
+{"id":"f13_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/projectDb_old

@@ -0,0 +1,12 @@
+{"id":"f1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":4}
+{"id":"f1__________::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":1}
+{"id":"f11_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"f11_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"f11_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"f12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":3}
+{"id":"f12_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":0,"views":3}
+{"id":"f12_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"f12_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"f13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":8}
+{"id":"f13_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":0,"views":2}
+{"id":"f13_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/usageDb

@@ -0,0 +1,9 @@
+{"id":"dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":5}
+{"id":"doi_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"doi_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"doi_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"doi_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":6}
+{"id":"doi_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"doi_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"doi_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":10}
+{"id":"doi_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

dhp-workflows/dhp-aggregation/src/test/resources/eu/dnetlib/dhp/actionmanager/usagestats/usageDb_old

@@ -0,0 +1,12 @@
+{"id":"dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":4}
+{"id":"dedup_wf_001::53575dc69e9ace947e02d47ecd54a7a6","downloads":0,"views":1}
+{"id":"doi_________::17eda2ff77407538fbe5d3d719b9d1c0","downloads":0,"views":1}
+{"id":"doi_________::1d4dc08605fd0a2be1105d30c63bfea1","downloads":1,"views":3}
+{"id":"doi_________::2e3527822854ca9816f6dfea5bff61a8","downloads":1,"views":1}
+{"id":"doi_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":2,"views":3}
+{"id":"doi_________::3085e4c6e051378ca6157fe7f0430c1f","downloads":0,"views":3}
+{"id":"doi_________::33f710e6dd30cc5e67e35b371ddc33cf","downloads":0,"views":1}
+{"id":"doi_________::39738ebf10654732dd3a7af9f24655f8","downloads":1,"views":3}
+{"id":"doi_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":1,"views":8}
+{"id":"doi_________::3c3b65f07c1a06c7894397eda1d11bbf","downloads":0,"views":2}
+{"id":"doi_________::4938a71a884dd481d329657aa543b850","downloads":0,"views":3}

pom.xml

@@ -801,7 +801,7 @@
 		<mockito-core.version>3.3.3</mockito-core.version>
 		<mongodb.driver.version>3.4.2</mongodb.driver.version>
 		<vtd.version>[2.12,3.0)</vtd.version>
-		<dhp-schemas.version>[2.12.1-patched]</dhp-schemas.version>
+		<dhp-schemas.version>[2.13.1-patched]</dhp-schemas.version>
 		<dnet-actionmanager-api.version>[4.0.3]</dnet-actionmanager-api.version>
 		<dnet-actionmanager-common.version>[6.0.5]</dnet-actionmanager-common.version>
 		<dnet-openaire-broker-common.version>[3.1.6]</dnet-openaire-broker-common.version>