Accommodating for HopsUtil renaming to Hops.

Author: tkakantousis

README.md

@@ -14,7 +14,7 @@ mvn package
 Generates a jar for each module which can then either be used to create HopsWorks jobs (Spark/Flink) or execute Hive queries remotely.
 
 # Helper Libraries
-Hops Examples makes use of **HopsUtil**, a set of Java and Python libraries which provide developers with tools that make programming on Hops easy. *HopsUtil* is automatically made available to all Jobs and Notebooks, without the user having to explicitely import it. Detailed documentation on HopsUtil is available [here](https://github.com/hopshadoop/hops-util). 
+Hops Examples makes use of **Hops**, a set of Java and Python libraries which provide developers with tools that make programming on Hops easy. *Hops* is automatically made available to all Jobs and Notebooks, without the user having to explicitely import it. Detailed documentation on Hops is available [here](https://github.com/hopshadoop/hops-util). 
 
 # Spark
 ## Structured Streaming with Kafka and HopsFS
@@ -35,7 +35,7 @@ Usage: <type>(producer|consumer)
 Data consumed is be default persisted to the `Resources` dataset of the Project where the job is running.
 
 ### Avro Records
-`StructuredStreamingKafka.java` generates *String <key,value>* pairs which are converted by **HopsUtil** into Avro records and serialized into bytes. Similarly, during consuming from a Kafka source, messages are deserialized into Avro records. **The default Avro schema used is the following**:
+`StructuredStreamingKafka.java` generates *String <key,value>* pairs which are converted by **Hops** into Avro records and serialized into bytes. Similarly, during consuming from a Kafka source, messages are deserialized into Avro records. **The default Avro schema used is the following**:
 
 ```json
 {
@@ -140,7 +140,7 @@ For Avro schemas with more fields, the application's SourceFunction should use a
 
 2. For examples on customizing logging for Flink jobs on HopsWorks see [here](https://github.com/hopshadoop/hops-kafka-examples/tree/master/examples-flink).
 
-3. *StreamingExample* makes use of [here](https://github.com/hopshadoop/hops-util). When building this project, HopsUtil is automatically included in the assembled jar file.
+3. *StreamingExample* makes use of [here](https://github.com/hopshadoop/hops-util). When building this project, Hops is automatically included in the assembled jar file.
 
 
 ## Job Logging

flink/src/main/java/io/hops/examples/flink/kafka/FlinkKafkaExample.java

@@ -2,7 +2,7 @@
 
 import io.hops.util.HopsConsumer;
 import io.hops.util.HopsProducer;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import java.util.Map;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -75,15 +75,15 @@ public static void main(String[] args) throws Exception {
     // execute the program
     env.execute("Streaming Iteration Example");
 
-    HopsUtil.setup(HopsUtil.getFlinkKafkaProps(args[args.length - 1]));
+    Hops.setup(Hops.getFlinkKafkaProps(args[args.length - 1]));
     if (args[1].equalsIgnoreCase("producer")) {
       Configuration hdConf = new Configuration();
       Path hdPath = new org.apache.hadoop.fs.Path(args[2]);
       FileSystem hdfs = hdPath.getFileSystem(hdConf);
       final FSDataOutputStream stream = hdfs.create(hdPath);
       stream.write("My first Flink program on Hops!".getBytes());
 
-      HopsProducer hopsKafkaProducer = HopsUtil.getHopsProducer(args[0]);
+      HopsProducer hopsKafkaProducer = Hops.getHopsProducer(args[0]);
       Map<String, String> message;
       for (int i = 0; i < 30; i++) {
         message = new HashMap<>();
@@ -96,7 +96,7 @@ public static void main(String[] args) throws Exception {
       stream.close();
       hopsKafkaProducer.close();
     } else {
-      final HopsConsumer hopsKafkaConsumer = HopsUtil.getHopsConsumer(args[0]);
+      final HopsConsumer hopsKafkaConsumer = Hops.getHopsConsumer(args[0]);
       Thread t = new Thread() {
         @Override
         public void run() {

flink/src/main/java/io/hops/examples/flink/kafka/StreamingExample.java

@@ -1,7 +1,7 @@
 package io.hops.examples.flink.kafka;
 
 import io.hops.util.Constants;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import java.util.Arrays;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -41,7 +41,7 @@ public static void main(String[] args) throws Exception {
 
     ////////////////////////////////////////////////////////////////////////////
     //Hopsworks utility method to automatically set parameters for Kafka
-    HopsUtil.setup(HopsUtil.getFlinkKafkaProps(parameterTool.get(
+    Hops.setup(Hops.getFlinkKafkaProps(parameterTool.get(
         Constants.KAFKA_FLINK_PARAMS)));
     ////////////////////////////////////////////////////////////////////////////
     if (parameterTool.get("type").equalsIgnoreCase("producer")) {
@@ -75,8 +75,8 @@ public void cancel() {
           });
 
       // write data into Kafka
-      for (String topic : HopsUtil.getTopics()) {
-        messageStream.addSink(HopsUtil.getFlinkProducer(topic));
+      for (String topic : Hops.getTopics()) {
+        messageStream.addSink(Hops.getFlinkProducer(topic));
       }
       env.execute("Write into Kafka example");
     } else {
@@ -91,8 +91,8 @@ public void cancel() {
       env.getConfig().setGlobalJobParameters(ParameterTool.fromArgs(
           Arrays.copyOf(args, args.length - 2)));
 
-      for (String topic : HopsUtil.getTopics()) {
-        DataStream<String> messageStream = env.addSource(HopsUtil.
+      for (String topic : Hops.getTopics()) {
+        DataStream<String> messageStream = env.addSource(Hops.
             getFlinkConsumer(topic));
         String dateTimeBucketerFormat = "yyyy-MM-dd--HH";
         if (parameterTool.has("sink_path")) {

spark/src/main/java/io/hops/examples/spark/WorkflowExample.java

@@ -1,7 +1,7 @@
 package io.hops.examples.spark;
 
 import io.hops.util.exceptions.CredentialsNotFoundException;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import io.hops.util.WorkflowManager;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -25,7 +25,7 @@ public static void main(String[] args)
 
     SparkSession spark = SparkSession
         .builder()
-        .appName(HopsUtil.getJobName())
+        .appName(Hops.getJobName())
         .getOrCreate();
 
     //if Start job with given ID

spark/src/main/java/io/hops/examples/spark/kafka/SimpleKafkaProcess.java

@@ -2,7 +2,7 @@
 
 import io.hops.util.HopsConsumer;
 import io.hops.util.HopsProducer;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -69,7 +69,7 @@ public static void main(String[] args) throws Exception {
             "yourtopic")});
 
       //Produce Kafka messages to topic
-      HopsProducer hopsKafkaProducer = HopsUtil.getHopsProducer(topic);
+      HopsProducer hopsKafkaProducer = Hops.getHopsProducer(topic);
 
       Map<String, String> message;
       int i = 0;
@@ -86,7 +86,7 @@ public static void main(String[] args) throws Exception {
       hopsKafkaProducer.close();
     } else {
       //Consume kafka messages from topic
-      HopsConsumer hopsKafkaConsumer = HopsUtil.getHopsConsumer(topic);
+      HopsConsumer hopsKafkaConsumer = Hops.getHopsConsumer(topic);
       //Keep thread alive
       //THIS WILL CAUSE THE JOB TO HANG. USER HAS TO MANUALLY STOP THE JOB.
       while (true) {

spark/src/main/java/io/hops/examples/spark/kafka/StreamingExample.java

@@ -4,7 +4,7 @@
 import com.twitter.bijection.Injection;
 import io.hops.util.exceptions.CredentialsNotFoundException;
 import io.hops.util.HopsProducer;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import io.hops.util.exceptions.SchemaNotFoundException;
 import io.hops.util.spark.SparkConsumer;
 import io.hops.util.spark.SparkProducer;
@@ -59,7 +59,7 @@ public final class StreamingExample {
   private static final Pattern SPACE = Pattern.compile(" ");
   //Get HopsWorks Kafka Utility instance
   private static final Map<String, Injection<GenericRecord, byte[]>> recordInjections
-      = HopsUtil.getRecordInjections();
+      = Hops.getRecordInjections();
 
   public static void main(final String[] args) throws Exception {
     if (args.length < 1) {
@@ -71,7 +71,7 @@ public static void main(final String[] args) throws Exception {
 
     final String type = args[0];
     // Create context with a 2 ; batch interval
-    Set<String> topicsSet = new HashSet<>(HopsUtil.getTopics());
+    Set<String> topicsSet = new HashSet<>(Hops.getTopics());
     SparkConf sparkConf = new SparkConf().setAppName("StreamingExample");
     final List<HopsProducer> sparkProducers = new ArrayList<>();
 
@@ -83,7 +83,7 @@ public static void main(final String[] args) throws Exception {
           @Override
           public void run() {
             try {
-              SparkProducer sparkProducer = HopsUtil.getSparkProducer(topic);
+              SparkProducer sparkProducer = Hops.getSparkProducer(topic);
               sparkProducers.add(sparkProducer);
               Map<String, String> message = new HashMap<>();
               int i = 0;
@@ -102,16 +102,16 @@ public void run() {
           }
         }.start();
       }//Keep application running
-      HopsUtil.shutdownGracefully(jsc);
+      Hops.shutdownGracefully(jsc);
     } else {
       JavaStreamingContext jssc = new JavaStreamingContext(sparkConf,
           Durations.seconds(2));
       //Use applicationId for sink folder
       final String appId = jssc.sparkContext().getConf().getAppId();
 
       //Get consumer groups
-      List<String> consumerGroups = HopsUtil.getConsumerGroups();
-      SparkConsumer consumer = HopsUtil.getSparkConsumer(jssc, topicsSet);
+      List<String> consumerGroups = Hops.getConsumerGroups();
+      SparkConsumer consumer = Hops.getSparkConsumer(jssc, topicsSet);
       // Create direct kafka stream with topics
       JavaInputDStream<ConsumerRecord<String, byte[]>> messages = consumer.
           createDirectStream();
@@ -186,7 +186,7 @@ public void call(JavaPairRDD<String, Integer> rdd, Time time) throws
        */
       // Start the computation
       jssc.start();
-      HopsUtil.shutdownGracefully(jssc);
+      Hops.shutdownGracefully(jssc);
     }
     for (HopsProducer hopsProducer : sparkProducers) {
       hopsProducer.close();

spark/src/main/java/io/hops/examples/spark/kafka/StreamingKafkaElastic.java

@@ -2,7 +2,7 @@
 
 import com.google.common.base.Strings;
 import io.hops.util.exceptions.CredentialsNotFoundException;
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import io.hops.util.exceptions.WorkflowManagerException;
 import io.hops.util.spark.SparkConsumer;
 import java.io.BufferedReader;
@@ -55,7 +55,7 @@ public final class StreamingKafkaElastic {
 
   public static void main(final String[] args) throws Exception {
 
-    SparkConf sparkConf = new SparkConf().setAppName(HopsUtil.getJobName());
+    SparkConf sparkConf = new SparkConf().setAppName(Hops.getJobName());
     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(10));
 
     //Use applicationId for sink folder
@@ -64,8 +64,8 @@ public static void main(final String[] args) throws Exception {
     //Get consumer groups
     Properties props = new Properties();
     props.put("value.deserializer", StringDeserializer.class.getName());
-    props.put("client.id", HopsUtil.getJobName());
-    SparkConsumer consumer = HopsUtil.getSparkConsumer(jssc, props);
+    props.put("client.id", Hops.getJobName());
+    SparkConsumer consumer = Hops.getSparkConsumer(jssc, props);
     //Store processed offsets
 
     // Create direct kafka stream with topics
@@ -103,7 +103,7 @@ public void call(JavaRDD<LogEntryFilebeat> rdd, Time time) throws Exception {
 
           DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyyMMdd");
           LocalDate localDate = LocalDate.now();
-          row.write().mode(SaveMode.Append).parquet("/Projects/" + HopsUtil.getProjectName() + "/" + dataset + "/Logs-"
+          row.write().mode(SaveMode.Append).parquet("/Projects/" + Hops.getProjectName() + "/" + dataset + "/Logs-"
               + dtf.format(localDate));
         }
       }
@@ -118,7 +118,7 @@ public void call(JavaRDD<LogEntryFilebeat> rdd, Time time) throws Exception {
      */
     // Start the computation
     jssc.start();
-    HopsUtil.shutdownGracefully(jssc);
+    Hops.shutdownGracefully(jssc);
   }
 
   private static JSONObject parser(String logType, String line, String appId)
@@ -165,9 +165,9 @@ private static JSONObject parser(String logType, String line, String appId)
       index.put("file", jsonLog.getString("source"));
       index.put("application", appId);
       index.put("host", jsonLog.getJSONObject("beat").getString("hostname"));
-      index.put("project", HopsUtil.getProjectName());
+      index.put("project", Hops.getProjectName());
       index.put("method", "?");
-      index.put("jobname", HopsUtil.getJobName());
+      index.put("jobname", Hops.getJobName());
     } else if (logType.equalsIgnoreCase("custom")) {
       //Catch an error getting log attributes and set default ones in this case
       //Example message
@@ -228,9 +228,9 @@ private static JSONObject parser(String logType, String line, String appId)
       }
       index.put("application", appId);
       index.put("host", jsonLog.getJSONObject("beat").getString("hostname"));
-      index.put("project", HopsUtil.getProjectName());
+      index.put("project", Hops.getProjectName());
       index.put("method", "?");
-      index.put("jobname", HopsUtil.getJobName());
+      index.put("jobname", Hops.getJobName());
 
       if (jsonLog.getString("source").contains("shrek")) {
         index.put("location", "59.32,18.06");
@@ -239,15 +239,15 @@ private static JSONObject parser(String logType, String line, String appId)
       }
       if (!Strings.isNullOrEmpty(priority) && priority.equalsIgnoreCase("TRACE")) {
         LOG.log(Level.INFO, "Sending email");
-        HopsUtil.getWorkflowManager().sendEmail("[email protected]", "Error message received", timestamp + " :: " + message);
+        Hops.getWorkflowManager().sendEmail("[email protected]", "Error message received", timestamp + " :: " + message);
       }
 
     }
     URL obj;
     HttpURLConnection conn = null;
     BufferedReader br = null;
     try {
-      obj = new URL("http://" + HopsUtil.getElasticEndPoint() + "/" + HopsUtil.getProjectName().toLowerCase()
+      obj = new URL("http://" + Hops.getElasticEndPoint() + "/" + Hops.getProjectName().toLowerCase()
           + "/logs");
       conn = (HttpURLConnection) obj.openConnection();
       conn.setDoOutput(true);

spark/src/main/java/io/hops/examples/spark/kafka/StreamingLogs.java

@@ -1,6 +1,6 @@
 package io.hops.examples.spark.kafka;
 
-import io.hops.util.HopsUtil;
+import io.hops.util.Hops;
 import io.hops.util.exceptions.SchemaNotFoundException;
 import io.hops.util.spark.SparkConsumer;
 import java.io.IOException;
@@ -49,7 +49,7 @@ public final class StreamingLogs {
 
   public static void main(final String[] args) throws Exception {
 
-    SparkConf sparkConf = new SparkConf().setAppName(HopsUtil.getJobName());
+    SparkConf sparkConf = new SparkConf().setAppName(Hops.getJobName());
     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
 
     //Use applicationId for sink folder
@@ -58,8 +58,8 @@ public static void main(final String[] args) throws Exception {
     //Get consumer groups
     Properties props = new Properties();
     props.put("value.deserializer", StringDeserializer.class.getName());
-    props.put("client.id", HopsUtil.getJobName());
-    SparkConsumer consumer = HopsUtil.getSparkConsumer(jssc, props);
+    props.put("client.id", Hops.getJobName());
+    SparkConsumer consumer = Hops.getSparkConsumer(jssc, props);
     //Store processed offsets
 
     // Create direct kafka stream with topics
@@ -97,7 +97,7 @@ public void call(JavaRDD<NamenodeLogEntry> rdd, Time time) throws
         Dataset<Row> row = sparkSession.createDataFrame(rdd, NamenodeLogEntry.class);
         if (!rdd.isEmpty()) {
           row.write().mode(SaveMode.Append).
-              parquet("/Projects/" + HopsUtil.getProjectName() + "/Resources/LogAnalysis");
+              parquet("/Projects/" + Hops.getProjectName() + "/Resources/LogAnalysis");
         }
       }
     });
@@ -111,7 +111,7 @@ public void call(JavaRDD<NamenodeLogEntry> rdd, Time time) throws
      */
     // Start the computation
     jssc.start();
-    HopsUtil.shutdownGracefully(jssc);
+    Hops.shutdownGracefully(jssc);
   }
 
   private static JSONObject parser(String line, String appId) {
@@ -152,8 +152,8 @@ private static JSONObject parser(String line, String appId) {
     index.put("timestamp", timestamp);
     index.put("application", appId);
     index.put("host", jsonLog.getJSONObject("beat").getString("hostname"));
-    index.put("project", HopsUtil.getProjectName());
-    index.put("jobname", HopsUtil.getJobName());
+    index.put("project", Hops.getProjectName());
+    index.put("jobname", Hops.getJobName());
     if (jsonLog.getString("source").contains("/")) {
       index.put("file", jsonLog.getString("source").substring(jsonLog.getString("source").lastIndexOf("/") + 1));
     } else {