andreas.czerniak
/
BrBETA_dnet-hadoop
forked from D-Net/dnet-hadoop
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

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

This commit is contained in:
Claudio Atzori 2020-03-04 10:57:06 +01:00
parent 0233987603 086af63158
commit 6379f32466
3 changed files with 25 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/migration/step2/GenerateEntitiesApplication.java
View File

@ -53,18 +53,20 @@ public class GenerateEntitiesApplication {
final String dbUser = parser.get("postgresUser");
final String dbPassword = parser.get("postgresPassword");
final SparkSession spark = SparkSession
final Map<String, String> code2name = loadClassNames(dbUrl, dbUser, dbPassword);
try (final SparkSession spark = newSparkSession(parser); final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext())) {
final List<String> existingSourcePaths = Arrays.stream(sourcePaths.split(",")).filter(p -> exists(sc, p)).collect(Collectors.toList());
generateEntities(sc, code2name, existingSourcePaths, targetPath);
}
}
private static SparkSession newSparkSession(final ArgumentApplicationParser parser) {
return SparkSession
.builder()
.appName(GenerateEntitiesApplication.class.getSimpleName())
.master(parser.get("master"))
.getOrCreate();
final Map<String, String> code2name = loadClassNames(dbUrl, dbUser, dbPassword);
try (final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext())) {
final List<String> existingSourcePaths = Arrays.stream(sourcePaths.split(",")).filter(p -> exists(sc, p)).collect(Collectors.toList());
generateEntities(sc, code2name, existingSourcePaths, targetPath);
}
}
private static void generateEntities(final JavaSparkContext sc,

16
dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/migration/step3/DispatchEntitiesApplication.java
View File

@ -28,13 +28,7 @@ public class DispatchEntitiesApplication {
.getResourceAsStream("/eu/dnetlib/dhp/migration/dispatch_entities_parameters.json")));
parser.parseArgument(args);
final SparkSession spark = SparkSession
.builder()
.appName(DispatchEntitiesApplication.class.getSimpleName())
.master(parser.get("master"))
.getOrCreate();
try (final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext())) {
try (final SparkSession spark = newSparkSession(parser); final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext())) {
final String sourcePath = parser.get("sourcePath");
final String targetPath = parser.get("graphRawPath");
@ -50,6 +44,14 @@ public class DispatchEntitiesApplication {
}
}
private static SparkSession newSparkSession(final ArgumentApplicationParser parser) {
return SparkSession
.builder()
.appName(DispatchEntitiesApplication.class.getSimpleName())
.master(parser.get("master"))
.getOrCreate();
}
private static void processEntity(final JavaSparkContext sc, final Class<?> clazz, final String sourcePath, final String targetPath) {
final String type = clazz.getSimpleName().toLowerCase();

10
dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/migration/utils/DbClient.java
View File

@ -28,8 +28,8 @@ public class DbClient implements Closeable {
StringUtils.isNoneBlank(login, password) ? DriverManager.getConnection(address, login, password) : DriverManager.getConnection(address);
this.connection.setAutoCommit(false);
} catch (final Exception e) {
log.error(e.getClass().getName() + ": " + e.getMessage());
throw new RuntimeException(e);
log.error("Connection to postgresDB failed");
throw new RuntimeException("Connection to postgresDB failed", e);
}
log.info("Opened database successfully");
}
@ -44,10 +44,12 @@ public class DbClient implements Closeable {
consumer.accept(rs);
}
} catch (final SQLException e) {
throw new RuntimeException(e);
log.error("Error executing sql query: " + sql, e);
throw new RuntimeException("Error executing sql query", e);
}
} catch (final SQLException e1) {
throw new RuntimeException(e1);
log.error("Error preparing sql statement", e1);
throw new RuntimeException("Error preparing sql statement", e1);
}
}