Feat/prometheus script

.circleci/config.yml

@@ -0,0 +1,15 @@
+version: 2.1
+
+# Define the jobs we want to run for this project
+jobs:
+  build:
+    docker:
+      - image: openjdk:8-jdk-oraclelinux7
+    steps:
+      - run: echo "build job is not implemented"
+
+# Orchestrate our job run sequence
+workflows:
+  build:
+    jobs:
+      - build

.gitignore

@@ -28,3 +28,6 @@ project/plugins/project/
 
 # Node
 node_modules
+
+# Spark-ec2 boto
+tools/spark-ec2/lib

.gitmodules

@@ -0,0 +1,4 @@
+[submodule "tools/flintrock"]
+	path = tools/flintrock
+	url = git@github.com:chaordic/flintrock.git
+	branch = ignition_v1

README.md

@@ -8,4 +8,4 @@ It also provides many utilities for Spark jobs and Scala programs in general.
 It should be used inside a project as a submodule. See https://github.com/chaordic/ignition-template for an example. 
 
 # Getting started
-See http://monkeys.chaordic.com.br/start-using-spark-with-ignition/ for quick-start tutorial
+See [Start using Spark with Ignition!](http://monkeys.chaordic.com.br/2015/03/22/start-using-spark-with-ignition.html) for quick-start tutorial

build.sbt

@@ -2,37 +2,31 @@ name := "Ignition-Core"
 
 version := "1.0"
 
-scalaVersion := "2.10.4"
+scalaVersion := "2.11.12"
 
-scalacOptions ++= Seq("-unchecked", "-deprecation", "-feature", "-Xfatal-warnings")
-
-ideaExcludeFolders += ".idea"
-
-ideaExcludeFolders += ".idea_modules"
+scalacOptions ++= Seq("-unchecked", "-deprecation", "-feature", "-Xfatal-warnings", "-Xlint", "-Ywarn-dead-code", "-Xmax-classfile-name", "130")
 
 // Because we can't run two spark contexts on same VM
 parallelExecution in Test := false
 
-libraryDependencies += ("org.apache.spark" %% "spark-core" % "1.3.0" % "provided").exclude("org.apache.hadoop", "hadoop-client")
-
-libraryDependencies += ("org.apache.hadoop" % "hadoop-client" % "2.0.0-cdh4.7.1" % "provided")
+test in assembly := {}
 
-libraryDependencies += "com.github.nscala-time" %% "nscala-time" % "0.8.0"
+libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.3" % "provided"
 
-libraryDependencies += "org.scalatest" % "scalatest_2.10" % "2.0"
+libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.7.6" % "provided"
 
-libraryDependencies += "org.scalaj" %% "scalaj-http" % "0.3.16"
+libraryDependencies += "org.apache.hadoop" % "hadoop-aws" % "2.7.6" % "provided"
 
-libraryDependencies += "org.scalaz" %% "scalaz-core" % "7.0.6"
+libraryDependencies += "com.amazonaws" % "aws-java-sdk" % "1.7.4" % "provided"
 
-libraryDependencies += "com.github.scopt" %% "scopt" % "3.2.0"
+libraryDependencies += "org.scalaz" %% "scalaz-core" % "7.2.27"
 
-libraryDependencies += "net.java.dev.jets3t" % "jets3t" % "0.7.1"
+libraryDependencies += "com.github.scopt" %% "scopt" % "3.6.0"
 
-resolvers += "Akka Repository" at "http://repo.akka.io/releases/"
+libraryDependencies += "joda-time" % "joda-time" % "2.9.9"
 
-resolvers += "Sonatype OSS Releases" at "http://oss.sonatype.org/content/repositories/releases/"
+libraryDependencies += "org.joda" % "joda-convert" % "1.8.2"
 
-resolvers += "Cloudera Repository" at "https://repository.cloudera.com/artifactory/cloudera-repos/"
+libraryDependencies += "org.slf4j" % "slf4j-api" % "1.7.25"
 
-resolvers += Resolver.sonatypeRepo("public")
+libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.3"

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.5
+sbt.version=1.2.6

remote_hook.sh

@@ -1,5 +1,6 @@
 #!/bin/bash
 
+
 # We suppose we are in a subdirectory of the root project
 DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 
@@ -11,6 +12,8 @@ CONTROL_DIR="${5?Please give the Control Directory}"
 SPARK_MEM_PARAM="${6?Please give the Job Memory Size to use}"
 USE_YARN="${7?Please tell if we should use YARN (yes/no)}"
 NOTIFY_ON_ERRORS="${8?Please tell if we will notify on errors (yes/no)}"
+DRIVER_HEAP_SIZE="${9?Please tell driver heap size to use}"
+shift 9
 
 JOB_WITH_TAG=${JOB_NAME}.${JOB_TAG}
 JOB_CONTROL_DIR="${CONTROL_DIR}/${JOB_WITH_TAG}"
@@ -20,10 +23,18 @@ MY_USER=$(whoami)
 sudo mkdir -p "${JOB_CONTROL_DIR}"
 sudo chown $MY_USER "${JOB_CONTROL_DIR}"
 
-
 RUNNING_FILE="${JOB_CONTROL_DIR}/RUNNING"
+# This should be the first thing in the script to avoid the wait remote job thinking we died
 echo $$ > "${RUNNING_FILE}"
 
+
+
+# Let us read the spark home even when the image doesn't give us the permission
+sudo chmod o+rx /home/ec2-user
+sudo chmod -R o+rx /home/ec2-user/spark
+
+mkdir -p /media/tmp/spark-events
+
 notify_error_and_exit() {
     description="${1}"
     echo "Exiting because: ${description}"
@@ -48,6 +59,37 @@ on_trap_exit() {
     rm -f "${RUNNING_FILE}"
 }
 
+install_and_run_zeppelin() {
+    if [[ ! -d "zeppelin" ]]; then
+        wget "http://www-us.apache.org/dist/zeppelin/zeppelin-0.8.0/zeppelin-0.8.0-bin-all.tgz" -O zeppelin.tar.gz
+        mkdir zeppelin
+        tar xvzf zeppelin.tar.gz -C zeppelin --strip-components 1 > /tmp/zeppelin_install.log
+    fi
+    if [[ -f "zeppelin/bin/zeppelin.sh" ]]; then
+        export MASTER="${JOB_MASTER}"
+        export ZEPPELIN_PORT="8081"
+        export SPARK_HOME=$(get_first_present /root/spark /opt/spark ~/spark*/)
+        export SPARK_SUBMIT_OPTIONS="--jars ${JAR_PATH} --executor-memory ${SPARK_MEM_PARAM}"
+        zeppelin/bin/zeppelin.sh
+    else
+        notify_error_and_exit "Zeppelin installation not found"
+    fi
+}
+
+install_and_run_jupyter() {
+    sudo yum -y install python3 python3-pip
+    sudo pip3 install jupyter pandas boto3 matplotlib numpy sklearn scipy toree
+    export SPARK_HOME=$(get_first_present /root/spark /opt/spark ~/spark*/)
+    export HADOOP_HOME=$(get_first_present /root/hadoop /opt/hadoop ~/hadoop*/)
+    export SPARK_CONF_DIR="${SPARK_HOME}/conf"
+    export HADOOP_CONF_DIR="${HADOOP_HOME}/conf"
+    export JOB_MASTER=${MASTER:-spark://${SPARK_MASTER_HOST}:7077}
+    export PYSPARK_PYTHON=$(which python3)
+    export PYSPARK_DRIVER_PYTHON=$(which jupyter)
+    export PYSPARK_DRIVER_PYTHON_OPTS="notebook --allow-root --ip=${SPARK_MASTER_HOST} --no-browser --port=8888"
+    sudo $(which jupyter) toree install --spark_home="${SPARK_HOME}" --spark_opts="--master ${JOB_MASTER} --executor-memory ${SPARK_MEM_PARAM} --driver-memory ${DRIVER_HEAP_SIZE}"
+    ${SPARK_HOME}/bin/pyspark --master "${JOB_MASTER}" --executor-memory "${SPARK_MEM_PARAM}" --driver-memory "${DRIVER_HEAP_SIZE}"
+}
 
 trap "on_trap_exit" EXIT
 
@@ -58,12 +100,14 @@ MAIN_CLASS="ignition.jobs.Runner"
 
 cd "${DIR}" || notify_error_and_exit "Internal script error for job ${JOB_WITH_TAG}"
 
-JAR_PATH_SRC=$(echo "${DIR}"/*assembly*.jar)
+JAR_PATH_SRC=$(ls -t "${DIR}"/*assembly*.jar | head -1) # most recent jar
 JAR_PATH="${JOB_CONTROL_DIR}/Ignition.jar"
 
 cp ${JAR_PATH_SRC} ${JAR_PATH}
 
-export JOB_MASTER=${MASTER}
+# If no $MASTER, then build a url using $SPARK_MASTER_HOST
+export JOB_MASTER=${MASTER:-spark://${SPARK_MASTER_HOST}:7077}
+
 
 if [[ "${USE_YARN}" == "yes" ]]; then
     export YARN_MODE=true
@@ -73,14 +117,16 @@ if [[ "${USE_YARN}" == "yes" ]]; then
     export SPARK_WORKER_MEMORY=${SPARK_MEM_PARAM}
 fi
 
-
 if [[ "${JOB_NAME}" == "shell" ]]; then
-    export ADD_JARS=${JAR_PATH}
-    sudo -E ${SPARK_HOME}/bin/spark-shell || notify_error_and_exit "Execution failed for shell"
+    ${SPARK_HOME}/bin/spark-shell --master "${JOB_MASTER}" --jars ${JAR_PATH} --driver-memory "${DRIVER_HEAP_SIZE}" --driver-java-options "-Djava.io.tmpdir=/media/tmp -verbose:gc -XX:-PrintGCDetails -XX:+PrintGCTimeStamps" --executor-memory "${SPARK_MEM_PARAM}" || notify_error_and_exit "Execution failed for shell"
+elif [[ "${JOB_NAME}" == "zeppelin" ]]; then
+    install_and_run_zeppelin
+elif [[ "${JOB_NAME}" == "jupyter" ]]; then
+    install_and_run_jupyter
 else
     JOB_OUTPUT="${JOB_CONTROL_DIR}/output.log"
     tail -F "${JOB_OUTPUT}" &
-    sudo -E "${SPARK_HOME}/bin/spark-submit" --master "${JOB_MASTER}" --driver-memory 25000M --driver-java-options "-Djava.io.tmpdir=/mnt -verbose:gc -XX:-PrintGCDetails -XX:+PrintGCTimeStamps" --class "${MAIN_CLASS}" ${JAR_PATH} "${JOB_NAME}" --runner-date "${JOB_DATE}" --runner-tag "${JOB_TAG}" --runner-user "${JOB_USER}" --runner-master "${JOB_MASTER}" --runner-executor-memory "${SPARK_MEM_PARAM}" >& "${JOB_OUTPUT}" || notify_error_and_exit "Execution failed for job ${JOB_WITH_TAG}"
+    ${SPARK_HOME}/bin/spark-submit --master "${JOB_MASTER}" --driver-memory "${DRIVER_HEAP_SIZE}" --driver-java-options "-Djava.io.tmpdir=/media/tmp -verbose:gc -XX:-PrintGCDetails -XX:+PrintGCTimeStamps " --class "${MAIN_CLASS}" ${JAR_PATH} "${JOB_NAME}" --runner-date "${JOB_DATE}" --runner-tag "${JOB_TAG}" --runner-user "${JOB_USER}" --runner-master "${JOB_MASTER}" --runner-executor-memory "${SPARK_MEM_PARAM}" "$@" >& "${JOB_OUTPUT}" || notify_error_and_exit "Execution failed for job ${JOB_WITH_TAG}"
 fi
 
 touch "${JOB_CONTROL_DIR}/SUCCESS"

src/main/scala/ignition/core/jobs/CoreJobRunner.scala

@@ -1,21 +1,31 @@
 package ignition.core.jobs
 
-import org.apache.spark.{SparkConf, SparkContext}
-import org.joda.time.{DateTimeZone, DateTime}
+import org.apache.spark.SparkContext
+import org.apache.spark.sql.SparkSession
+import org.joda.time.{DateTime, DateTimeZone}
+import org.slf4j.{Logger, LoggerFactory}
 
-import scala.util.Try
+import scala.concurrent.Future
 
 object CoreJobRunner {
 
+  val logger: Logger = LoggerFactory.getLogger(getClass)
+
   case class RunnerContext(sparkContext: SparkContext,
+                           sparkSession: SparkSession,
                            config: RunnerConfig)
 
 
   // Used to provide contextual logging
   def setLoggingContextValues(config: RunnerConfig): Unit = {
-    org.slf4j.MDC.put("setupName", config.setupName)
-    org.slf4j.MDC.put("tag", config.tag)
-    org.slf4j.MDC.put("user", config.user)
+    try { // yes, this may fail but we don't want everything to shut down
+      org.slf4j.MDC.put("setupName", config.setupName)
+      org.slf4j.MDC.put("tag", config.tag)
+      org.slf4j.MDC.put("user", config.user)
+    } catch {
+      case e: Throwable =>
+        // cry
+    }
   }
 
   case class RunnerConfig(setupName: String = "nosetup",
@@ -24,7 +34,7 @@ object CoreJobRunner {
                           user: String = "nouser",
                           master: String = "local[*]",
                           executorMemory: String = "2G",
-                          additionalArgs: Map[String, String] = Map.empty)
+                          extraArgs: Map[String, String] = Map.empty)
 
   def runJobSetup(args: Array[String], jobsSetups: Map[String, (CoreJobRunner.RunnerContext => Unit, Map[String, String])], defaultSparkConfMap: Map[String, String]) {
     val parser = new scopt.OptionParser[RunnerConfig]("Runner") {
@@ -49,8 +59,8 @@ object CoreJobRunner {
         c.copy(executorMemory = x)
       }
 
-      opt[(String, String)]('w', "runner-with-arg") unbounded() action { (x, c) =>
-        c.copy(additionalArgs = c.additionalArgs ++ Map(x))
+      opt[(String, String)]('w', "runner-extra") unbounded() action { (x, c) =>
+        c.copy(extraArgs = c.extraArgs ++ Map(x))
       }
     }
 
@@ -65,27 +75,39 @@ object CoreJobRunner {
       val appName = s"${config.setupName}.${config.tag}"
 
 
-      val sparkConf = new SparkConf()
-      sparkConf.set("spark.executor.memory", config.executorMemory)
+      val builder = SparkSession.builder
+      builder.config("spark.executor.memory", config.executorMemory)
+
+      builder.config("spark.eventLog.dir", "file:///media/tmp/spark-events")
+
+      builder.master(config.master)
+      builder.appName(appName)
 
-      sparkConf.setMaster(config.master)
-      sparkConf.setAppName(appName)
-
-      defaultSparkConfMap.foreach { case (k, v) => sparkConf.set(k, v) }
+      builder.config("spark.hadoop.mapred.output.committer.class", classOf[DirectOutputCommitter].getName())
 
-      jobConf.foreach { case (k, v) => sparkConf.set(k, v) }
+      defaultSparkConfMap.foreach { case (k, v) => builder.config(k, v) }
+
+      jobConf.foreach { case (k, v) => builder.config(k, v) }
 
       // Add logging context to driver
       setLoggingContextValues(config)
-
-      val sc = new SparkContext(sparkConf)
 
+      try {
+        builder.enableHiveSupport()
+      } catch {
+        case t: Throwable => logger.warn("Failed to enable HIVE support", t)
+      }
+
+      val session = builder.getOrCreate()
+
+      val sc = session.sparkContext
       // Also try to propagate logging context to workers
       // TODO: find a more efficient and bullet-proof way
       val configBroadCast = sc.broadcast(config)
+
       sc.parallelize(Range(1, 2000), numSlices = 2000).foreachPartition(_ => setLoggingContextValues(configBroadCast.value))
 
-      val context = RunnerContext(sc, config)
+      val context = RunnerContext(sc, session, config)
 
       try {
         jobSetup.apply(context)
@@ -94,8 +116,14 @@ object CoreJobRunner {
           t.printStackTrace()
           System.exit(1) // force exit of all threads
       }
-      Try { sc.stop() }
-      System.exit(0) // force exit of all threads
+
+      import scala.concurrent.ExecutionContext.Implicits.global
+      Future {
+        // If everything is fine, the system will shut down without the help of this thread and YARN will report success
+        // But sometimes it gets stuck, then it's necessary to use the force, but this may finish the job as failed on YARN
+        Thread.sleep(30 * 1000)
+        System.exit(0) // force exit of all threads
+      }
     }
   }
 }