Merge branch 'stable_ids' of https://code-repo.d4science.org/D-Net/dnet-hadoop into stable_ids

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/MigrateHdfsMdstoresApplication.java

@@ -52,9 +52,8 @@ public class MigrateHdfsMdstoresApplication extends AbstractMigrationApplication
 	public static void main(final String[] args) throws Exception {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
 			IOUtils
-				.toString(
-					MigrateHdfsMdstoresApplication.class
-						.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/migrate_hdfs_mstores_parameters.json")));
+				.toString(MigrateHdfsMdstoresApplication.class
+					.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/migrate_hdfs_mstores_parameters.json")));
 		parser.parseArgument(args);
 
 		final Boolean isSparkSessionManaged = Optional
@@ -94,14 +93,21 @@ public class MigrateHdfsMdstoresApplication extends AbstractMigrationApplication
 			.filter(p -> HdfsSupport.exists(p, sc.hadoopConfiguration()))
 			.toArray(size -> new String[size]);
 
-		spark
-			.read()
-			.parquet(validPaths)
-			.map((MapFunction<Row, String>) r -> enrichRecord(r), Encoders.STRING())
-			.toJavaRDD()
-			.mapToPair(xml -> new Tuple2<>(new Text(UUID.randomUUID() + ":" + type), new Text(xml)))
-			// .coalesce(1)
-			.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, GzipCodec.class);
+		if (validPaths.length > 0) {
+			spark
+				.read()
+				.parquet(validPaths)
+				.map((MapFunction<Row, String>) r -> enrichRecord(r), Encoders.STRING())
+				.toJavaRDD()
+				.mapToPair(xml -> new Tuple2<>(new Text(UUID.randomUUID() + ":" + type), new Text(xml)))
+				// .coalesce(1)
+				.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, GzipCodec.class);
+		} else {
+			spark.emptyDataFrame()
+				.toJavaRDD()
+				.mapToPair(xml -> new Tuple2<>(new Text(), new Text()))
+				.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, GzipCodec.class);
+		}
 	}
 
 	private static String enrichRecord(final Row r) {