D-Net
/
dnet-hadoop
14
5
Fork 11
Code Issues 14 Pull Requests 12 Projects 2 Releases 2 Wiki Activity

changed the way to save the results on filesystem

This commit is contained in:
Miriam Baglioni 2020-04-11 16:47:34 +02:00
parent 87f802821e
commit 8f12292daa
1 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java
View File

@ -100,7 +100,7 @@ public class SparkCountryPropagationJob2 {
broadcast_datasourcecountryassoc).as(Encoders.bean(ResultCountrySet.class)); broadcast_datasourcecountryassoc).as(Encoders.bean(ResultCountrySet.class));
if(writeUpdates){ if(writeUpdates){
writeUpdates(potentialUpdates.toJavaRDD(), outputPath + "/update_" + resultType); writeUpdates(potentialUpdates, outputPath + "/update_" + resultType);
} }
if(saveGraph){ if(saveGraph){
@ -147,9 +147,13 @@ public class SparkCountryPropagationJob2 {
log.info("Saving graph table to path: {}", outputPath); log.info("Saving graph table to path: {}", outputPath);
//log.info("number of saved recordsa: {}", new_table.count()); //log.info("number of saved recordsa: {}", new_table.count());
new_table new_table
.toJavaRDD() .toJSON()
.map(r -> OBJECT_MAPPER.writeValueAsString(r)) .write()
.saveAsTextFile(outputPath , GzipCodec.class); .option("compression", "gzip")
.text(outputPath);
// .toJavaRDD()
// .map(r -> OBJECT_MAPPER.writeValueAsString(r))
// .saveAsTextFile(outputPath , GzipCodec.class);
} }
@ -219,9 +223,14 @@ public class SparkCountryPropagationJob2 {
.map(value -> OBJECT_MAPPER.readValue(value, DatasourceCountry.class), Encoders.bean(DatasourceCountry.class)); .map(value -> OBJECT_MAPPER.readValue(value, DatasourceCountry.class), Encoders.bean(DatasourceCountry.class));
} }
private static void writeUpdates(JavaRDD<ResultCountrySet> potentialUpdates, String outputPath){ private static void writeUpdates(Dataset<ResultCountrySet> potentialUpdates, String outputPath){
potentialUpdates.map(u -> OBJECT_MAPPER.writeValueAsString(u)) potentialUpdates
.saveAsTextFile(outputPath, GzipCodec.class); .toJSON()
.write()
.option("compression", "gzip")
.text(outputPath);
// map(u -> OBJECT_MAPPER.writeValueAsString(u))
// .saveAsTextFile(outputPath, GzipCodec.class);
} }