simplified import procedure, maps the infospace as hive tables

dhp-workflows/dhp-graph-mapper/pom.xml

@@ -32,11 +32,6 @@
             <version>1.0.4-SNAPSHOT</version>
         </dependency>
 
-        <dependency>
-            <groupId>de.javakaffee</groupId>
-            <artifactId>kryo-serializers</artifactId>
-        </dependency>
-
     </dependencies>
 
 

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/graph/SparkGraphImporterJob.java

@@ -3,11 +3,11 @@ package eu.dnetlib.dhp.graph;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import eu.dnetlib.dhp.application.ArgumentApplicationParser;
 import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
 import org.apache.spark.sql.SparkSession;
 import scala.Tuple2;
 
@@ -19,29 +19,27 @@ public class SparkGraphImporterJob {
         parser.parseArgument(args);
         final SparkSession spark = SparkSession
                 .builder()
-                .appName("ImportGraph")
+                .appName(SparkGraphImporterJob.class.getSimpleName())
                 .master(parser.get("master"))
+                .config("hive.metastore.uris", parser.get("hive.metastore.uris"))
+                .enableHiveSupport()
                 .getOrCreate();
+
         final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
         final String inputPath = parser.get("sourcePath");
-        final String outputPath = parser.get("targetPath");
-
-        final String filter = parser.get("filter");
 
         // Read the input file and convert it into RDD of serializable object
-        final JavaRDD<Tuple2<String, String>> inputRDD = sc.sequenceFile(inputPath, Text.class, Text.class)
-                .map(item -> new Tuple2<>(item._1.toString(), item._2.toString()));
-
         GraphMappingUtils.types.forEach((name, clazz) -> {
-            if (StringUtils.isNotBlank(filter) || filter.toLowerCase().contains(name)) {
+            final JavaRDD<Tuple2<String, String>> inputRDD = sc.sequenceFile(inputPath + "/" + name, Text.class, Text.class)
+                    .map(item -> new Tuple2<>(item._1.toString(), item._2.toString()));
             spark.createDataset(inputRDD
                     .filter(s -> s._1().equals(clazz.getName()))
                     .map(Tuple2::_2)
                     .map(s -> new ObjectMapper().readValue(s, clazz))
                     .rdd(), Encoders.bean(clazz))
                     .write()
-                        .save(outputPath + "/" + name);
+                    .mode(SaveMode.Overwrite)
-            }
+                    .saveAsTable("openaire." + name);
         });
 
     }

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/graph/input_graph_parameters.json

@@ -1,6 +1,5 @@
 [
   {"paramName":"mt",  "paramLongName":"master",             "paramDescription": "should be local or yarn",                                  "paramRequired": true},
   {"paramName":"s",   "paramLongName":"sourcePath",         "paramDescription": "the path of the sequencial file to read",                  "paramRequired": true},
-  {"paramName":"f",   "paramLongName":"filter",             "paramDescription": "csv of typology of dataframe to be generated",             "paramRequired": false},
+  {"paramName":"h",   "paramLongName":"hive.metastore.uris","paramDescription": "the hive metastore uris",                                  "paramRequired": true}
-  {"paramName":"t",   "paramLongName":"targetPath",         "paramDescription": "the path where store DataFrames on HDFS",                  "paramRequired": true}
 ]

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/graph/oozie_app/config-default.xml

@@ -16,7 +16,7 @@
         <value>spark2</value>
     </property>
     <property>
-        <name>filter</name>
+        <name>hive.metastore.uris</name>
-        <value></value>
+        <value>thrift://iis-cdh5-test-m3.ocean.icm.edu.pl:9083</value>
     </property>
 </configuration>

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/graph/oozie_app/workflow.xml

@@ -4,10 +4,6 @@
             <name>sourcePath</name>
             <description>the source path</description>
         </property>
-        <property>
-            <name>targetPath</name>
-            <description>the target path</description>
-        </property>
         <property>
             <name>sparkDriverMemory</name>
             <description>memory for driver process</description>
@@ -20,10 +16,6 @@
             <name>sparkExecutorCores</name>
             <description>number of cores used by single executor</description>
         </property>
-        <property>
-            <name>filter</name>
-            <description>csv string to filter the entities to generate</description>
-        </property>
     </parameters>
 
     <start to="MapGraphIntoDataFrame"/>
@@ -41,12 +33,10 @@
             <name>MapGraphIntoDataFrame</name>
             <class>eu.dnetlib.dhp.graph.SparkGraphImporterJob</class>
             <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
-            <spark-opts>--executor-memory ${sparkExecutorMemory} --executor-cores ${sparkExecutorCores} --driver-memory=${sparkDriverMemory} --conf spark.extraListeners="com.cloudera.spark.lineage.NavigatorAppListener" --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener"</spark-opts>
+            <spark-opts>--executor-memory ${sparkExecutorMemory} --executor-cores ${sparkExecutorCores} --driver-memory=${sparkDriverMemory} --conf spark.extraListeners="com.cloudera.spark.lineage.NavigatorAppListener" --conf spark.sql.queryExecutionListeners="com.cloudera.spark.lineage.NavigatorQueryListener" --conf spark.sql.warehouse.dir="/user/hive/warehouse"</spark-opts>
             <arg>-mt</arg> <arg>yarn-cluster</arg>
             <arg>--sourcePath</arg><arg>${sourcePath}</arg>
-            <arg>--targetPath</arg><arg>${targetPath}</arg>
+            <arg>--hive.metastore.uris</arg><arg>${hive.metastore.uris}</arg>
-            <arg>--filter</arg><arg>${filter}</arg>
-
         </spark>
         <ok to="End"/>
         <error to="Kill"/>

pom.xml

@@ -164,12 +164,6 @@
                 <version>1.1.6</version>
             </dependency>
 
-            <dependency>
-                <groupId>de.javakaffee</groupId>
-                <artifactId>kryo-serializers</artifactId>
-                <version>0.45</version>
-            </dependency>
-
             <dependency>
                 <groupId>net.schmizz</groupId>
                 <artifactId>sshj</artifactId>