Apache Spark/scala中的Log4J Kafka功能

Question

嗨，我正在尝试从一群执行者那里登录到一个Kafka主题，他们使用ApacheSpark和Log4J以及KafkaAppender。我可以使用基本的文件附加器与执行者登录，但不能登录到Kafka。

这是我的log4j.properties我为此定制的：

log4j.rootLogger=INFO, console, KAFKA, file

log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n


log4j.appender.KAFKA=org.apache.kafka.log4jappender.KafkaLog4jAppender
log4j.appender.KAFKA.topic=test2
log4j.appender.KAFKA.name=localhost
log4j.appender.KAFKA.host=localhost
log4j.appender.KAFKA.port=9092
log4j.appender.KAFKA.brokerList=localhost:9092
log4j.appender.KAFKA.compressionType=none
log4j.appender.KAFKA.requiredNumAcks=0
log4j.appender.KAFKA.syncSend=true
log4j.appender.KAFKA.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L %% - %m%n



log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=log4j-application.log
log4j.appender.file.MaxFileSize=5MB
log4j.appender.file.MaxBackupIndex=10
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

这是我的代码（到目前为止）。我试图传递一个记录器定义，以便每个执行者都能得到一个副本，但我不知道为什么它不会传到Kafka：

import org.apache.log4j._
import org.apache.spark._
import org.apache.spark.rdd.RDD
import java.io._
import org.apache.kafka.log4jappender.KafkaLog4jAppender

class Mapper(n: Int) extends Serializable{
  @transient lazy val suplogger: Logger = Logger.getLogger("myLogger")

  def doSomeMappingOnDataSetAndLogIt(rdd: RDD[Int]): RDD[String] =
    rdd.map{ i =>
      val sparkConf: SparkConf =new org.apache.spark.SparkConf()
      logger.setLevel((Level) Level.ALL)
      suplogger.warn(sparkConf.toDebugString)
      val pid = Integer.parseInt(new File("/proc/self").getCanonicalFile().getName());
      suplogger.warn("--------------------")
      suplogger.warn("mapping: " + i)
      val supIterator = new scala.collection.JavaConversions.JEnumerationWrapper(suplogger.getAllAppenders())
      suplogger.warn("List is " + supIterator.toList)
      suplogger.warn("Num of list is: " + supIterator.size)

      //(i + n).toString
      "executor pid = "+pid + "debug string: " + sparkConf.toDebugString.size
    }
}

object Mapper {
  def apply(n: Int): Mapper = new Mapper(n)
}

object HelloWorld {
  def main(args: Array[String]): Unit = {
    println("sup")
    println("yo")
    val log = LogManager.getRootLogger
    log.setLevel(Level.WARN)
    val nameIterator = new scala.collection.JavaConversions.JEnumerationWrapper(log.getAllAppenders())
    println(nameIterator.toList)

    val conf = new SparkConf().setAppName("demo-app")
    val sc = new SparkContext(conf)
    log.warn(conf.toDebugString)
    val pid = Integer.parseInt(new File("/proc/self").getCanonicalFile().getName());
    log.warn("--------------------")
    log.warn("IP: "+java.net.InetAddress.getLocalHost() +" PId: "+pid)

    log.warn("Hello demo")

    val data = sc.parallelize(1 to 100, 10)

    val mapper = Mapper(1)

    val other = mapper.doSomeMappingOnDataSetAndLogIt(data)

    other.collect()

    log.warn("I am done")
  }

}

下面是日志文件的一些示例输出：

2017-01-25 06:29:15 WARN  myLogger:19 - spark.driver.port=54335
2017-01-25 06:29:15 WARN  myLogger:21 - --------------------
2017-01-25 06:29:15 WARN  myLogger:23 - mapping: 1
2017-01-25 06:29:15 WARN  myLogger:25 - List is List()
2017-01-25 06:29:15 WARN  myLogger:26 - Num of list is: 0
2017-01-25 06:29:15 WARN  myLogger:19 - spark.driver.port=54335
2017-01-25 06:29:15 WARN  myLogger:21 - --------------------
2017-01-25 06:29:15 WARN  myLogger:23 - mapping: 2
2017-01-25 06:29:15 WARN  myLogger:25 - List is List()
2017-01-25 06:29:15 WARN  myLogger:26 - Num of list is: 0
2017-01-25 06:29:15 WARN  myLogger:19 - spark.driver.port=54335
2017-01-25 06:29:15 WARN  myLogger:21 - --------------------

谢谢你们的帮助，如果你们有什么需要的话，请告诉我！

这是spark提交命令的副本

spark-submit \
    --deploy-mode client \
    --files spark_test/mylogger.props \
    --packages "com.databricks:spark-csv_2.10:1.4.0,org.apache.kafka:kafka-log4j-appender:0.10.1.1" \
    --num-executors 4 \
    --executor-cores 1 \
    --driver-java-options "-Dlog4j.configuration=file:///home/mapr/spark_test/mylogger.props" \
    --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file:///home/mapr/spark_test/mylogger.props" \
    --class "HelloWorld" helloworld.jar

Answer 1

您的问题是没有正确地将spark_test/mylogger.props文件传递给执行程序。

无论如何，您都需要使用执行程序的file上传文件。

spark-submit \
    --deploy-mode client \
    --driver-java-options "-Dlog4j.configuration=file:/home/mapr/spark_test/mylogger.props \
    --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file:mylogger.props" \
    --files /home/mapr/spark_test/mylogger.props \
    ...

spark-submit \
    --deploy-mode cluster \
    --conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=file:mylogger.props" \
    --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=file:mylogger.props" \
    --files /home/mapr/spark_test/mylogger.props \
    ...

查看我关于配置Spark日志记录的完整帖子：
https://stackoverflow.com/a/55596389/1549135

有关Spark Kafka appender的更多详细信息：
https://stackoverflow.com/a/58883911/1549135

Answer 2

我知道问题是什么了。我没有部署到集群，我只是在客户端模式下部署。说实话，我不知道为什么当我被发送到集群时，这会起作用。

我使用的是MapR沙盒虚拟机https://www.mapr.com/products/mapr-sandbox-hadoop

如果有人能解释为什么客户机/集群在这里发挥了作用，我将非常感激！