【发布时间】：2017-06-13 07:40:59
【问题描述】：
您好，我正在尝试使用带有 Log4J 和 Kafka-Appender 添加的 Apache Spark 从一堆执行程序中登录到 Kafka 主题。我可以使用基本的 File Appender 与执行者一起登录，但不能登录到 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
这是我的代码（到目前为止）。我试图传递一个记录器定义，以便每个执行者都得到一个副本，但我不知道为什么它不会去卡夫卡：
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-submit 命令的副本
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
【问题讨论】：
-
您在哪里运行 Spark 作业？在 YARN 集群中？如果是，那么您的代理列表（本地主机）可能有问题。你在 stdout/stderr 看到什么了吗？
-
我实际上发现了问题所在！我会跟进的
-
@TobiSH 我想通了，但我不确定为什么我的解决方案有效
标签： scala logging apache-spark log4j apache-kafka