Scala中的Apache Spark日志记录

Question

我正在寻找一种解决方案，能够在Apache Spark节点上执行代码时记录额外的数据，这有助于调查稍后在执行过程中可能出现的一些问题。尝试使用传统的解决方案，例如com。打字安全。鳞片法。LazyLogging失败，因为无法在Apache Spark这样的分布式环境中序列化日志实例。

我已经研究了这个问题，现在我发现的解决方案是使用org.apache.spark.Logging特性如下：

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

然而，日志特性似乎不是Apache Spark的永久解决方案，因为它被标记为@DeveloperApi，并且类留档提到：

这可能会在未来的版本中更改或删除。

我想知道——它们是否有我可以使用的已知日志记录解决方案，并允许我在Apache Spark节点上执行RDD时记录数据？

@稍后编辑：下面的一些评论建议使用Log4J。我尝试过使用Log4J，但在使用Scala类（而不是Scala对象）中的记录器时仍然存在问题。以下是我的完整代码：

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()
 }

}

我看到的例外是：

线程"main"中的异常org.apache.spark.Spark异常：任务不可序列化-

Answer 1

如果需要在map、filter或其他RDD函数之前和之后执行一些代码，请尝试使用mapPartition，其中显式传递基础迭代器。

例子：

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

变成：

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

每个基本的RDD函数总是使用mapPartition实现。

确保显式地处理分区程序，而不是松动它：请参见Scaladoc，保护分区参数，这对性能至关重要。

Answer 2

使用Log4j 2。x、 core logger已可序列化。问题解决了。

Jira讨论：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"

Answer 3

您可以使用Akhil在
https://www.mail-archive.com/user@火花。阿帕奇。org/msg29010。html。我自己用过，效果很好。

Akhil Das Mon，2015年5月25日08:20:40-0700
试着这样做：

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)
}