From 36f8f2ea66412f909cd8983cc0718b0ab6e10641 Mon Sep 17 00:00:00 2001 From: Claudio Atzori Date: Wed, 25 Mar 2020 14:16:06 +0100 Subject: [PATCH] master set to 'yarn' in spark actions, removed path to rawSet from the dedup scan workflow --- .../dnetlib/dhp/dedup/SparkCreateSimRels.java | 24 +++++-------------- .../dedup/consistency/oozie_app/workflow.xml | 4 ++-- .../dhp/dedup/createSimRels_parameters.json | 6 ----- .../dhp/dedup/scan/oozie_app/workflow.xml | 9 ++----- 4 files changed, 10 insertions(+), 33 deletions(-) diff --git a/dhp-workflows/dhp-dedup/src/main/java/eu/dnetlib/dhp/dedup/SparkCreateSimRels.java b/dhp-workflows/dhp-dedup/src/main/java/eu/dnetlib/dhp/dedup/SparkCreateSimRels.java index 8c0efdcad..0fc72db1e 100644 --- a/dhp-workflows/dhp-dedup/src/main/java/eu/dnetlib/dhp/dedup/SparkCreateSimRels.java +++ b/dhp-workflows/dhp-dedup/src/main/java/eu/dnetlib/dhp/dedup/SparkCreateSimRels.java @@ -43,22 +43,17 @@ public class SparkCreateSimRels implements Serializable { //read oozie parameters final String graphBasePath = parser.get("graphBasePath"); final String isLookUpUrl = parser.get("isLookUpUrl"); - final String rawSet = parser.get("rawSet"); final String actionSetId = parser.get("actionSetId"); final String workingPath = parser.get("workingPath"); System.out.println(String.format("graphBasePath: '%s'", graphBasePath)); System.out.println(String.format("isLookUpUrl: '%s'", isLookUpUrl)); - System.out.println(String.format("rawSet: '%s'", rawSet)); System.out.println(String.format("actionSetId: '%s'", actionSetId)); System.out.println(String.format("workingPath: '%s'", workingPath)); try (SparkSession spark = getSparkSession(parser)) { final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext()); - //create empty sequenceFile for the accumulation - JavaRDD> simRel = sc.emptyRDD(); - //for each dedup configuration for (DedupConfig dedupConf: DedupUtility.getConfigurations(isLookUpUrl, actionSetId)) { final String entity = dedupConf.getWf().getEntityType(); @@ -83,23 +78,16 @@ public class SparkCreateSimRels implements Serializable { .write() .mode("overwrite") .save(DedupUtility.createSimRelPath(workingPath, actionSetId, subEntity)); - - if (rawSet != null) { - //create atomic actions - JavaRDD> newSimRels = relationsRDD - .map(this::createSequenceFileRow); - - simRel = simRel.union(newSimRels); - } } - - if (rawSet != null) - simRel.mapToPair(r -> r) - .saveAsHadoopFile(rawSet, Text.class, Text.class, SequenceFileOutputFormat.class, GzipCodec.class); } - } + /** + * Utility method used to create an atomic action from a Relation object + * @param relation input relation + * @return A tuple2 with [id, json serialization of the atomic action] + * @throws JsonProcessingException + */ public Tuple2 createSequenceFileRow(Relation relation) throws JsonProcessingException { ObjectMapper mapper = new ObjectMapper(); diff --git a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/consistency/oozie_app/workflow.xml b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/consistency/oozie_app/workflow.xml index d3121ea77..d481a6cfb 100644 --- a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/consistency/oozie_app/workflow.xml +++ b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/consistency/oozie_app/workflow.xml @@ -65,7 +65,7 @@ --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener" --conf spark.sql.warehouse.dir="/user/hive/warehouse" - -mtyarn-cluster + -mtyarn --i${graphBasePath} --w${workingPath} --o${dedupGraphPath} @@ -92,7 +92,7 @@ --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener" --conf spark.sql.warehouse.dir="/user/hive/warehouse" - -mtyarn-cluster + -mtyarn --i${graphBasePath} --o${dedupGraphPath} --w${workingPath} diff --git a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/createSimRels_parameters.json b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/createSimRels_parameters.json index 9eb08a29b..8cffa86dc 100644 --- a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/createSimRels_parameters.json +++ b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/createSimRels_parameters.json @@ -23,12 +23,6 @@ "paramDescription": "the base path of the raw graph", "paramRequired": true }, - { - "paramName": "o", - "paramLongName": "rawSet", - "paramDescription": "the raw set to be saved (full path)", - "paramRequired": false - }, { "paramName": "w", "paramLongName": "workingPath", diff --git a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/scan/oozie_app/workflow.xml b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/scan/oozie_app/workflow.xml index abd152857..e2c7f425b 100644 --- a/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/scan/oozie_app/workflow.xml +++ b/dhp-workflows/dhp-dedup/src/main/resources/eu/dnetlib/dhp/dedup/scan/oozie_app/workflow.xml @@ -4,10 +4,6 @@ graphBasePath the raw graph base path - - rawSet - the output directory in the targetPath - isLookUpUrl the address of the lookUp service @@ -58,7 +54,6 @@ - yarn @@ -101,7 +96,7 @@ --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener" --conf spark.sql.warehouse.dir="/user/hive/warehouse" - -mtyarn-cluster + -mtyarn --i${graphBasePath} --w${workingPath} --la${isLookUpUrl} @@ -129,7 +124,7 @@ --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener" --conf spark.sql.warehouse.dir="/user/hive/warehouse" - -mtyarn-cluster + -mtyarn --i${graphBasePath} --w${workingPath} --la${isLookUpUrl}