dimitris.pierrakos
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
0
Fork 0
Code Pull Requests Activity

moved the version as it was before the change

This commit is contained in:
Miriam Baglioni 2023-05-09 10:48:34 +02:00
parent 99ac5bab46
commit 8c05f49665
1 changed files with 20 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/resulttocommunityfromorganization/PrepareResultCommunitySet.java
View File

@ -72,33 +72,28 @@ public class PrepareResultCommunitySet {
String outputPath, String outputPath,
OrganizationMap organizationMap) { OrganizationMap organizationMap) {
Dataset<Relation> relationAffiliation = readPath(spark, inputPath, Relation.class) Dataset<Relation> relation = readPath(spark, inputPath, Relation.class);
.filter( relation.createOrReplaceTempView("relation");
(FilterFunction<Relation>) r -> !r.getDataInfo().getDeletedbyinference() &&
r.getRelClass().equalsIgnoreCase(ModelConstants.HAS_AUTHOR_INSTITUTION));
Dataset<Relation> relationOrganization = readPath(spark, inputPath, Relation.class) String query = "SELECT result_organization.source resultId, result_organization.target orgId, org_set merges "
.filter( + "FROM (SELECT source, target "
(FilterFunction<Relation>) r -> !r.getDataInfo().getDeletedbyinference() && + " FROM relation "
r.getRelClass().equalsIgnoreCase(ModelConstants.MERGES)); + " WHERE datainfo.deletedbyinference = false "
+ " AND lower(relClass) = '"
+ ModelConstants.HAS_AUTHOR_INSTITUTION.toLowerCase()
+ "') result_organization "
+ "LEFT JOIN (SELECT source, collect_set(target) org_set "
+ " FROM relation "
+ " WHERE datainfo.deletedbyinference = false "
+ " AND lower(relClass) = '"
+ ModelConstants.MERGES.toLowerCase()
+ "' "
+ " GROUP BY source) organization_organization "
+ "ON result_organization.target = organization_organization.source ";
Dataset<ResultOrganizations> result_organizationset = relationAffiliation Dataset<ResultOrganizations> result_organizationset = spark
.joinWith( .sql(query)
relationOrganization, .as(Encoders.bean(ResultOrganizations.class));
relationAffiliation.col("target").equalTo(relationOrganization.col("source")),
"left")
.groupByKey((MapFunction<Tuple2<Relation, Relation>, String>) t2 -> t2._2().getSource(), Encoders.STRING())
.mapGroups((MapGroupsFunction<String, Tuple2<Relation, Relation>, ResultOrganizations>) (k, it) -> {
ResultOrganizations rOrgs = new ResultOrganizations();
rOrgs.setOrgId(k);
Tuple2<Relation, Relation> first = it.next();
rOrgs.setResultId(first._1().getSource());
ArrayList<String> merges = new ArrayList<>();
merges.add(first._2().getTarget());
it.forEachRemaining(t -> merges.add(t._2().getTarget()));
rOrgs.setMerges(merges);
return rOrgs;
}, Encoders.bean(ResultOrganizations.class));
result_organizationset result_organizationset
.map(mapResultCommunityFn(organizationMap), Encoders.bean(ResultCommunityList.class)) .map(mapResultCommunityFn(organizationMap), Encoders.bean(ResultCommunityList.class))