Spark Logging + Change log level runtime
This blog explains a way to change the custom logging level in a spark on the fly without restarting the application. I will be using log4j2 logger in the sample code used.
Sample scala application:
package com.jaal
import org.apache.logging.log4j.LogManager
import org.apache.spark.sql.SparkSession
object ReadTest {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("ReadTest").master("yarn").getOrCreate()
spark.sparkContext.setLogLevel("WARN")
import spark.implicits._
val sc = spark.sparkContext;
import scala.io.Source
val linesRead = spark.sparkContext.longAccumulator("LinesRead")
var i = 0
while(i < 25) {
val loglevelFile = "/tmp/loglevel.out"
var lev = ""
for (line <- Source.fromFile(loglevelFile).getLines) {
lev = line
}
val loglevelSeverityFile = "/tmp/loglevelSeverity.out"
var levSer = 400
for (line <- Source.fromFile(loglevelSeverityFile).getLines) {
levSer = line.toInt
}
sc.parallelize(Seq(""), 100).foreachPartition(x => {
import org.apache.logging.log4j.{LogManager, Level}
import org.apache.logging.log4j.core.config.Configurator
Configurator.setRootLevel(Level.forName(lev, levSer))
Configurator.setLevel("streamingtest", Level.forName(lev, levSer))
val log = LogManager.getRootLogger()
log.info("STARTING EXECUTOR INFO LOGGING")
log.warn("STARTING EXECUTOR WARN LOGGING")
log.error("STARTING EXECUTOR ERROR LOGGING")
log.fatal("STARTING EXECUTOR FATAL LOGGING")
log.debug("STARTING EXECUTOR DEBUG LOGGING")
})
val in = spark.read.text("/tmp/test.txt")
val inp = in.repartition(100)
inp.show(false)
linesRead.add(inp.count())
val logger = LogManager.getLogger("streamingtest")
import org.apache.spark.sql.functions.udf
val uppUdf = udf { s: String => {
val logger = LogManager.getLogger("streamingtest")
logger.debug("Inside upp")
logger.info("Inside upp")
logger.warn("Inside upp")
logger.error("Inside upp")
logger.fatal("Inside upp")
s.toUpperCase
}
}
val upC = inp.withColumn("upper",uppUdf($"value"))
upC.show(10)
logger.info("AlwinInfo")
logger.debug("AlwinDebug")
logger.error("AlwinError")
logger.fatal("AlwinFatal")
Thread.sleep(10000)
i = i+1
}
println(linesRead.value)
println(linesRead.avg)
}
}
The above application is good enough to explain and test the concept. All what the above application does is to read a file and convert the contents of it into upper case. This logic is running inside a loop that will terminate after executing 25 times. This is just to give us some breathing time to change the log level and verify that it has taken into effect.
The magic happens inside the following code snippet:
sc.parallelize(Seq(""), 100).foreachPartition(x => {
import org.apache.logging.log4j.{LogManager, Level}
import org.apache.logging.log4j.core.config.Configurator
Configurator.setRootLevel(Level.forName(lev, levSer))
Configurator.setLevel("streamingtest", Level.forName(lev, levSer))
val log = LogManager.getRootLogger()
log.info("STARTING EXECUTOR INFO LOGGING")
log.warn("STARTING EXECUTOR WARN LOGGING")
log.error("STARTING EXECUTOR ERROR LOGGING")
log.fatal("STARTING EXECUTOR FATAL LOGGING")
log.debug("STARTING EXECUTOR DEBUG LOGGING")
})
The above code snippet will set log level to ‘lev’ and ‘levSer’ across all the executor where this action is executed. You might have also noticed that the number of partitions for the above action is 100. You will need to adjust this number based on the number of executors that is running in the cluster. A safe bet is to set it to 2 times the number of executors you have so that it will run at least once in all of them. ‘lev’ and ‘levSer’ is read from a local file as shown below:
val loglevelFile = "/tmp/loglevel.out"
var lev = ""
for (line <- Source.fromFile(loglevelFile).getLines) {
lev = line
}
val loglevelSeverityFile = "/tmp/loglevelSeverity.out"
var levSer = 400
for (line <- Source.fromFile(loglevelSeverityFile).getLines) {
levSer = line.toInt
}
So in this case, in order to change the log level of the application, you just need to change the level inside these files. More details about log4j2 log level can be found here: https://logging.apache.org/log4j/2.x/manual/customloglevels.html
