spark executorのカスタムlog4jアペンダー

Question

私は、すべてのログをApache Kafkaに転送するために、spark executor内にカスタムlog4jアペンダーを使用しようとしています。

問題は、log4jのは、アペンダとfatjarのクラスローダーが登録される前に初期化され、私は次で終わるされている：

log4j:ERROR Could not instantiate class [kafka.producer.KafkaLog4jAppender]. 
java.lang.ClassNotFoundException: kafka.producer.KafkaLog4jAppender 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:372) 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:361) 
    at java.security.AccessController.doPrivileged(Native Method) 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:360) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424) 
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357) 
    at java.lang.Class.forName0(Native Method) 
    at java.lang.Class.forName(Class.java:260) 
    at org.apache.log4j.helpers.Loader.loadClass(Loader.java:198) 
    at org.apache.log4j.helpers.OptionConverter.instantiateByClassName(OptionConverter.java:327) 
    at org.apache.log4j.helpers.OptionConverter.instantiateByKey(OptionConverter.java:124) 
    at org.apache.log4j.PropertyConfigurator.parseAppender(PropertyConfigurator.java:785) 
    at org.apache.log4j.PropertyConfigurator.parseCategory(PropertyConfigurator.java:768) 
    at org.apache.log4j.PropertyConfigurator.configureRootCategory(PropertyConfigurator.java:648) 
    at org.apache.log4j.PropertyConfigurator.doConfigure(PropertyConfigurator.java:514) 
    at org.apache.log4j.PropertyConfigurator.doConfigure(PropertyConfigurator.java:580) 
    at org.apache.log4j.helpers.OptionConverter.selectAndConfigure(OptionConverter.java:526) 
    at org.apache.log4j.LogManager.<clinit>(LogManager.java:127) 
    at org.apache.spark.Logging$class.initializeLogging(Logging.scala:122) 
    at org.apache.spark.Logging$class.initializeIfNecessary(Logging.scala:107) 
    at org.apache.spark.Logging$class.log(Logging.scala:51) 
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.log(CoarseGrainedExecutorBackend.scala:126) 
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.run(CoarseGrainedExecutorBackend.scala:137) 
    at org.apache.spark.executor.CoarseGrainedExecutorBackend$.main(CoarseGrainedExecutorBackend.scala:235) 
    at org.apache.spark.executor.CoarseGrainedExecutorBackend.main(CoarseGrainedExecutorBackend.scala) 
log4j:ERROR Could not instantiate appender named "KAFKA". 
2015-09-29 13:10:43 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO akka.event.slf4j.Slf4jLogger: Slf4jLogger started 
2015-09-29 13:10:43 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO Remoting: Starting remoting 
2015-09-29 13:10:43 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO Remoting: Remoting started; listening on addresses :[akka.tcp://driverPropsFetcher@gin3.dev:36918] 
2015-09-29 13:10:43 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO Remoting: Remoting now listens on addresses: [akka.tcp://driverPropsFetcher@gin3.dev:36918] 
2015-09-29 13:10:44 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO akka.remote.RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon. 
2015-09-29 13:10:44 [driverPropsFetcher-akka.actor.default-dispatcher-4] INFO akka.remote.RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports. 
2015-09-29 13:10:44 [sparkExecutor-akka.actor.default-dispatcher-3] INFO akka.event.slf4j.Slf4jLogger: Slf4jLogger started 
2015-09-29 13:10:44 [sparkExecutor-akka.actor.default-dispatcher-2] INFO Remoting: Starting remoting 
2015-09-29 13:10:44 [sparkExecutor-akka.actor.default-dispatcher-2] INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkExecutor@gin3.dev:40067] 
2015-09-29 13:10:44 [sparkExecutor-akka.actor.default-dispatcher-2] INFO Remoting: Remoting now listens on addresses: [akka.tcp://sparkExecutor@gin3.dev:40067] 
2015-09-29 13:10:44 [driverPropsFetcher-akka.actor.default-dispatcher-5] INFO Remoting: Remoting shut down 
.... 

問題はここのようです：https://github.com/apache/spark/blob/v1.3.1/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala#L126

は、任意の使いやすいがありますこれを解決する方法？現在、Spark 1.3.xを使用しています。

おかげ

デビッド

Answer 1

ログアウトして余分なjarを送信し、ユーザークラスパスの前にロードすることを終了しました。

LOG_JAR="${THISDIR}/../lib/logging.jar" 
spark-submit ...... \ 
    --files "${LOG4J_CONF},${LOG_JAR}" \ 
    --conf "spark.driver.extraJavaOptions=-Dlog4j.configuration=`basename ${LOG4J_CONF}`" \ 
    --conf "spark.driver.extraClassPath=`basename ${LOG_JAR}`" \ 
    --conf "spark.executor.extraJavaOptions=-Dlog4j.configuration=`basename ${LOG4J_CONF}`" \ 
    --conf "spark.executor.extraClassPath=`basename ${LOG_JAR}`" \ 
    ... 

https://issues.apache.org/jira/browse/SPARK-10881?filter=-2

Answer 2

kafka.producer.KafkaLog4jAppenderはカフカのHadoopの-プロデューサーです。

この依存関係を追加して修正することができます。

<dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>hadoop-producer</artifactId> 
    <version>0.8.0</version> 
</dependency> 

Answer 3

同じ問題に直面していた、私は私のために働いていたものを掲載します、それはカフカ0.9で変更さKafkaLog4jAppenderクラスのパッケージ名が判明、ここで私がやったことで、POMに次の依存関係を追加しました

<dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka-log4j-appender</artifactId> 
     <version>0.9.0.0</version> 
    </dependency> 

と

log4j.appender.KAFKA=kafka.producer.KafkaLog4jAppender 

から私のlog4j.propertiesを変更

〜

log4j.appender.KAFKA=org.apache.kafka.log4jappender.KafkaLog4jAppender