forked from D-Net/dnet-hadoop
removed unnecessary variable and unused method
This commit is contained in:
parent
5b3ed70808
commit
17680296b9
|
@ -8,6 +8,7 @@ import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
import eu.dnetlib.dhp.schema.common.ModelSupport;
|
||||||
import org.apache.commons.io.IOUtils;
|
import org.apache.commons.io.IOUtils;
|
||||||
import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos;
|
import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos;
|
||||||
import org.apache.hadoop.io.Text;
|
import org.apache.hadoop.io.Text;
|
||||||
|
@ -115,7 +116,7 @@ public class SparkAtomicActionScoreJob implements Serializable {
|
||||||
"where dataInfo.deletedbyinference = false and pIde.qualifier.classid = '" + DOI + "'")
|
"where dataInfo.deletedbyinference = false and pIde.qualifier.classid = '" + DOI + "'")
|
||||||
.as(Encoders.bean(PreparedResult.class));
|
.as(Encoders.bean(PreparedResult.class));
|
||||||
|
|
||||||
Dataset<BipScore> tmp = bipScores
|
bipScores
|
||||||
.joinWith(
|
.joinWith(
|
||||||
preparedResult, bipScores.col("id").equalTo(preparedResult.col("value")),
|
preparedResult, bipScores.col("id").equalTo(preparedResult.col("value")),
|
||||||
"inner")
|
"inner")
|
||||||
|
@ -123,9 +124,7 @@ public class SparkAtomicActionScoreJob implements Serializable {
|
||||||
BipScore ret = value._1();
|
BipScore ret = value._1();
|
||||||
ret.setId(value._2().getId());
|
ret.setId(value._2().getId());
|
||||||
return ret;
|
return ret;
|
||||||
}, Encoders.bean(BipScore.class));
|
}, Encoders.bean(BipScore.class))
|
||||||
|
|
||||||
tmp
|
|
||||||
.groupByKey((MapFunction<BipScore, String>) value -> value.getId(), Encoders.STRING())
|
.groupByKey((MapFunction<BipScore, String>) value -> value.getId(), Encoders.STRING())
|
||||||
.mapGroups((MapGroupsFunction<String, BipScore, I>) (k, it) -> {
|
.mapGroups((MapGroupsFunction<String, BipScore, I>) (k, it) -> {
|
||||||
Result ret = inputClazz.newInstance();
|
Result ret = inputClazz.newInstance();
|
||||||
|
@ -143,18 +142,9 @@ public class SparkAtomicActionScoreJob implements Serializable {
|
||||||
aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
|
aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
|
||||||
new Text(OBJECT_MAPPER.writeValueAsString(aa))))
|
new Text(OBJECT_MAPPER.writeValueAsString(aa))))
|
||||||
.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class);
|
.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class);
|
||||||
;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public static Dataset<BipScore> getBipScoreDataset(Dataset<BipDeserialize> bipdeserialized) {
|
|
||||||
return bipdeserialized
|
|
||||||
.flatMap((FlatMapFunction<BipDeserialize, BipScore>) bip -> bip.keySet().stream().map(key -> {
|
|
||||||
BipScore bs = new BipScore();
|
|
||||||
bs.setId(key);
|
|
||||||
bs.setScoreList(bip.get(key));
|
|
||||||
return bs;
|
|
||||||
}).collect(Collectors.toList()).iterator(), Encoders.bean(BipScore.class));
|
|
||||||
}
|
|
||||||
|
|
||||||
private static List<Measure> getMeasure(BipScore value) {
|
private static List<Measure> getMeasure(BipScore value) {
|
||||||
return value
|
return value
|
||||||
|
|
Loading…
Reference in New Issue