D-Net
/
dnet-hadoop
14
5
Fork 11
Code Issues 14 Pull Requests 12 Projects 2 Releases 2 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 +3 -2
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 7dbdd4a0fe - Show all commits

5
dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/clean/country/CleanCountrySparkJob.java
View File

@ -10,6 +10,7 @@ import java.util.stream.Collectors;
import javax.swing.text.html.Option; import javax.swing.text.html.Option;
import eu.dnetlib.dhp.schema.oaf.utils.PidType;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.spark.SparkConf; import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FilterFunction; import org.apache.spark.api.java.function.FilterFunction;
@ -94,7 +95,6 @@ public class CleanCountrySparkJob implements Serializable {
List<String> hostedBy = spark List<String> hostedBy = spark
.read() .read()
.textFile(datasourcePath) .textFile(datasourcePath)
// .filter((FilterFunction<String>) ds -> !ds.equals(collectedfrom))
.collectAsList(); .collectAsList();
Dataset<T> res = spark Dataset<T> res = spark
@ -113,7 +113,8 @@ public class CleanCountrySparkJob implements Serializable {
if (r if (r
.getPid() .getPid()
.stream() .stream()
.anyMatch(p -> p.getQualifier().getClassid().equals("doi") && pidInParam(p.getValue(), verifyParam))) { .anyMatch(p -> p.getQualifier().getClassid()
.equals(PidType.doi) && pidInParam(p.getValue(), verifyParam))) {
r r
.setCountry( .setCountry(
r r