Add logback logger

README.md

@@ -2,6 +2,7 @@
 
 #Blog
 http://ashkrit.blogspot.com/2018/05/spark-microservices.html
+http://ashkrit.blogspot.sg/2018/05/custom-logs-in-apache-spark.html
 
 #How to Build
 

src/main/resources/logback.xml

@@ -0,0 +1,25 @@
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <!-- encoders are assigned the type
+             ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="METRICS" class="micro.logback.MetricsLogbackAppender"/>
+
+    <root level="info">
+        <appender-ref ref="STDOUT" />
+    </root>
+
+    <logger level="info" name="micro" additivity="true">
+        <appender-ref ref="METRICS" />
+    </logger>
+
+    <logger level="info" name="org.apache.spark.scheduler.DAGScheduler" additivity="true">
+        <appender-ref ref="METRICS" />
+    </logger>
+
+</configuration>

src/main/scala/micro/logback/MetricsLogbackAppender.scala

@@ -0,0 +1,30 @@
+package micro.logback
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.function.BiFunction
+
+import ch.qos.logback.classic.spi.ILoggingEvent
+import ch.qos.logback.core.UnsynchronizedAppenderBase
+
+class MetricsLogbackAppender extends UnsynchronizedAppenderBase[ILoggingEvent] {
+
+  override def append(e: ILoggingEvent) = {
+    //Send this message to elastic search or REST end point
+    messageCount.compute(Thread.currentThread().getName, mergeValue)
+    System.out.println(messageCount + " " + e)
+  }
+
+  val messageCount = new ConcurrentHashMap[String, AtomicInteger]()
+  val mergeValue = new BiFunction[String, AtomicInteger, AtomicInteger] {
+    def apply(key: String, currentValue: AtomicInteger) = {
+      val nextValue = currentValue match {
+        case null => new AtomicInteger(0)
+        case _ => currentValue
+      }
+      nextValue.incrementAndGet()
+      nextValue
+    }
+  }
+
+}

src/main/scala/micro/main/SparkFactory.scala

@@ -8,30 +8,33 @@ import org.slf4j.LoggerFactory
 
 object SparkFactory {
 
-   val _LOG = LoggerFactory.getLogger(this.getClass.getName)
+  val _LOG = LoggerFactory.getLogger(this.getClass.getName)
 
-  def newSparkSession(appName: String, master: String) = {
+  def newSparkSession(appName: String, master: String): SparkSession = {
     val sparkConf = new SparkConf().setAppName(appName).setMaster(master)
     val sparkSession = SparkSession.builder()
       .appName(appName)
       .config(sparkConf)
       .getOrCreate()
+
+    _LOG.info("Context details {}", sparkSession)
     sparkSession
+
   }
 
   def loadAsTable(context: SparkSession, fileName: String, table: String): SparkSession = {
     import context.implicits._
     val data = context.read.format("csv").option("header", "true").csv(fileName).as[Vehicle]
     data.printSchema()
+    _LOG.info("Schema details {}", data.schema.treeString)
     data.cache()
     data.createOrReplaceTempView(table)
     context
   }
 
 
   def filePath(filePath: String): String = {
-    val url = this.getClass.getResource("/logback.xml")
-    _LOG.info("Path {}" , url)
+    _LOG.info("Loading content from file {}", filePath)
     val fileUrl = this.getClass.getResource(filePath)
     Paths.get(fileUrl.toURI).toFile.getAbsolutePath
   }