Apache Spark logging within Scala

Question

I am looking for a solution to be able to log additional data when executing code on Apache Spark Nodes that could help investigate later some issues that might appear during execution. Trying to use a traditional solution like for example com.typesafe.scalalogging.LazyLogging fails because the log instance cannot be serialized on a distributed environment like Apache Spark.

I've investigated this problem and for now the solution that I found was to use the org.apache.spark.Logging trait like this :

class SparkExample with Logging {
  val someRDD = ...
  someRDD.map {
    rddElement => logInfo(s"$rddElement will be processed.")
    doSomething(rddElement)
  }
}

However it looks like the Logging trait is not a permanent solution for Apache Spark because it's marked as @DeveloperApi and the class documentation mentions:

This will likely be changed or removed in future releases.

I am wondering - are they any known logging solution that I can use and will allow me to log data when the RDDs are executed on Apache Spark nodes ?

@Later Edit : Some of the comments from below suggest to use Log4J. I've tried using Log4J but I'm still having issues when using logger from a Scala class (and not a Scala object). Here is my full code :

import org.apache.log4j.Logger
import org.apache.spark._

object Main {
 def main(args: Array[String]) {
  new LoggingTestWithRDD().doTest()
 }
}

class LoggingTestWithRDD extends Serializable {

  val log = Logger.getLogger(getClass.getName)

  def doTest(): Unit = {
   val conf = new SparkConf().setMaster("local[4]").setAppName("LogTest")
   val spark = new SparkContext(conf)

   val someRdd = spark.parallelize(List(1, 2, 3))
   someRdd.map {
     element =>
       log.info(s"$element will be processed")
       element + 1
    }
   spark.stop()
 }

}

The exception that I'm seeing is :

Exception in thread "main" org.apache.spark.SparkException: Task not serializable -> Caused by: java.io.NotSerializableException: org.apache.log4j.Logger

In addition to / as an alternative to logging, metrics may give you what you want: spark.apache.org/docs/latest/monitoring.html — michael
– michael, Commented Mar 23, 2015 at 17:19
@michael_n That's not correct. log4j and slf4j are different thing. — ben_frankly
– ben_frankly, Commented Aug 5, 2015 at 23:21

BAR · Accepted Answer · 2015-10-04 19:54:25Z

56

You can use Akhil's solution proposed in
https://www.mail-archive.com/[email protected]/msg29010.html. I have used by myself and it works.

Akhil Das Mon, 25 May 2015 08:20:40 -0700
Try this way:

object Holder extends Serializable {      
   @transient lazy val log = Logger.getLogger(getClass.getName)    
}


val someRdd = spark.parallelize(List(1, 2, 3)).foreach { element =>
   Holder.log.info(element)
}

edited Oct 4, 2015 at 19:54

BAR

17.5k28 gold badges108 silver badges213 bronze badges

answered May 26, 2015 at 8:42

florins

1,6651 gold badge18 silver badges35 bronze badges

Sign up to request clarification or add additional context in comments.

5 Comments

Rodrigo Del C. Andrade Over a year ago

I had a spark streaming Custom Receiver crapping NullPointering itself to death and spent a day pulling my beards. This was the solution. Thanks.

Knight71 Over a year ago

This seems to be work around. Say I want to enable a log from particular module. How to do that ?

Shilpa Over a year ago

I could not see any logs in the output logs. Could you suggest where I might be going wrong.

Justin Standard Over a year ago

I'm still having trouble with this exact approach within spark-jobserver. Anyone trying that configuration? Everything looks as though it should work, but I don't see any logging within the closure.

j pavan kumar Over a year ago

can you tell me where getLogger method is imported from

Ram Ghadiyaram · Accepted Answer · 2018-11-28 18:37:00Z

13

Use Log4j 2.x. The core logger has been made serializable. Problem solved.

Jira discussion: https://issues.apache.org/jira/browse/LOG4J2-801

"org.apache.logging.log4j" % "log4j-api" % "2.x.x"

"org.apache.logging.log4j" % "log4j-core" % "2.x.x"

"org.apache.logging.log4j" %% "log4j-api-scala" % "2.x.x"

edited Nov 28, 2018 at 18:37

Ram Ghadiyaram

29.4k16 gold badges102 silver badges133 bronze badges

answered Apr 25, 2017 at 17:35

Ryan Stack

1,3511 gold badge12 silver badges29 bronze badges

1 Comment

jAi Over a year ago

Can you please give complete implementation of this logging. like how you create log4j2.properties and how implemented in code.

ragazzojp · Accepted Answer · 2022-06-27 20:12:30Z

4

If you need some code to be executed before and after a map, filter or other RDD function, try to use mapPartition, where the underlying iterator is passed explicitely.

Example:

val log = ??? // this gets captured and produces serialization error
rdd.map { x =>
  log.info(x)
  x+1
}

Becomes:

rdd.mapPartition { it =>
  val log = ??? // this is freshly initialized in worker nodes
  it.map { x =>
    log.info(x)
    x + 1
  }
}

Every basic RDD function is implemented with a mapPartition.

Make sure to handle the partitioner explicitly and not to lose it: see Scaladoc, preservesPartitioning parameter, this is critical for performances.

edited Jun 27, 2022 at 20:12

answered May 29, 2018 at 11:35

ragazzojp

5595 silver badges17 bronze badges

Comments

Ram Ghadiyaram · Accepted Answer · 2018-11-28 18:38:04Z

2

This is an old post but I want to provide my working solution which I just got after struggling a lot and still can be useful for others:

I want to print rdd contents inside rdd.map function but getting Task Not Serializalable Error. This is my solution for this problem using scala static object which is extending java.io.Serializable:

import org.apache.log4j.Level

object MyClass extends Serializable{

val log = org.apache.log4j.LogManager.getLogger("name of my spark log")

log.setLevel(Level.INFO)

def main(args:Array[String])
{

rdd.map(t=>

//Using object's logger here

val log =MyClass.log

log.INFO("count"+rdd.count)
)
}

}

edited Nov 28, 2018 at 18:38

Ram Ghadiyaram

29.4k16 gold badges102 silver badges133 bronze badges

answered Jun 27, 2018 at 12:31

khushbu kanojia

2501 silver badge3 bronze badges

Comments

sparker · Accepted Answer · 2020-12-07 11:51:00Z

Making the logger transient and lazy does the trick

@transient lazy val log = Logger.getLogger(getClass.getName)

@transient will tell the spark to not serialize it for all executors and lazy will cause the instance to be created when it is first used. In other words each executor will have their own instance of the logger. Serializing the logger is not a good idea anyway even if you can.

Ofcourse anything you put in the map() closure will run on the executor so will be found in executor logs and not the driver logs. For custom log4j properties on the executors you need to add the log4j.properties to executor classpath and send your log4j.properties to the executors.

This can be done by adding the following args to your spark-submit command --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=./log4j.properties " --files ./log4j.properties There are other ways to do set these configs but this one is the most common.

bummi · Accepted Answer · 2015-05-26 11:06:04Z

1

val log = Logger.getLogger(getClass.getName),

You can use "log" to write logs . Also if you need change logger properties you need to have log4j.properties in /conf folder. By default we will have a template in that location.

edited May 26, 2015 at 11:06

bummi

27.4k14 gold badges67 silver badges105 bronze badges

answered Mar 23, 2015 at 17:11

Karthik

1,8211 gold badge14 silver badges21 bronze badges

4 Comments

Bogdan N Over a year ago

I've tried to use log4j but I am still having serialization issues when calling the logger variable from a class (not from a object) :

Exception in thread "main" org.apache.spark.SparkException: Task not serializable -> Caused by: java.io.NotSerializableException: org.apache.log4j.Logger

nuaavee Over a year ago

Simple solution: declare the log variable in local method scope.

Mikael Ståldal Over a year ago

What if you make "log" @transient ?

Ashkrit Sharma Over a year ago

Map partition comes to rescue for such thing. You can create logger in mappartitions function and use it. This technique is used for JDBC connection/ mq / Kafka producer.

TG Gowda · Accepted Answer · 2016-04-06 08:22:28Z

0

Here is my solution :

I am using SLF4j (with Log4j binding), in my base class of every spark job I have something like this:

import org.slf4j.LoggerFactory
val LOG = LoggerFactory.getLogger(getClass)

Just before the place where I use LOG in distributed functional code, I copy logger reference to a local constant.

val LOG = this.LOG

It worked for me!

answered Apr 6, 2016 at 8:22

TG Gowda

12.1k5 gold badges52 silver badges60 bronze badges

Collectives™ on Stack Overflow

Apache Spark logging within Scala

7 Answers 7

5 Comments

1 Comment

Comments

Comments

Comments

4 Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

7 Answers 7

5 Comments

1 Comment

Comments

Comments

Comments

4 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related