D-Net
/
dnet-hadoop
13
4
Fork 10
Code Issues 14 Pull Requests 9 Projects 2 Releases 1 Wiki Activity

[stats wf] indicators across stats dbs & updates in the org ids #248

Closed
dimitris.pierrakos wants to merge 1742 commits from beta into beta2master_sept_2022
pull from: beta
merge into: D-Net:beta2master_sept_2022
Conversation 3 Commits 1742 Files Changed 1525 +5 -5
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit f803530df6 - Show all commits

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

@ -75,17 +75,17 @@ public class SparkAtomicActionUsageJob implements Serializable {
isSparkSessionManaged, isSparkSessionManaged,
spark -> { spark -> {
removeOutputDir(spark, outputPath); removeOutputDir(spark, outputPath);
prepareData(dbname, spark, workingPath + "/usageDb", "usage_stats"); prepareData(dbname, spark, workingPath + "/usageDb", "usage_stats", "result_id");
prepareData(dbname, spark, workingPath + "/projectDb", "project_stats"); prepareData(dbname, spark, workingPath + "/projectDb", "project_stats", "id");
prepareData(dbname, spark, workingPath + "/datasourceDb", "datasource_stats"); prepareData(dbname, spark, workingPath + "/datasourceDb", "datasource_stats", "repositor_id");
writeActionSet(spark, workingPath, outputPath); writeActionSet(spark, workingPath, outputPath);
}); });
} }
private static void prepareData(String dbname, SparkSession spark, String workingPath, String tableName) { private static void prepareData(String dbname, SparkSession spark, String workingPath, String tableName, String attribute_name) {
spark spark
.sql( .sql(
"Select result_id, downloads, views " + "Select " + attribute_name + " as id, downloads, views " +
"from " + dbname + "." + tableName) "from " + dbname + "." + tableName)
.as(Encoders.bean(UsageStatsModel.class)) .as(Encoders.bean(UsageStatsModel.class))
.write() .write()