स्पार्क निष्पादक

Question

में कस्टम लॉग 4j एपेंडर मैं अपाचे काफ्का के सभी लॉग अग्रेषित करने के लिए स्पार्क निष्पादक के अंदर कस्टम log4j एपेंडर का उपयोग करने का प्रयास कर रहा हूं।

समस्या है, log4j से पहले appender साथ fatjar के classloader पंजीकृत हो जाता आरंभ नहीं हो जाता, तो मैं निम्नलिखित के साथ अंत:

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

वहाँ किसी भी आसान है इसे हल करने का तरीका? हम वर्तमान में स्पार्क 1.3.x का उपयोग कर रहे हैं।

धन्यवाद

डेविड

Answer 1

लॉगिंग डीपी के साथ अतिरिक्त जार जमा करना और उपयोगकर्ता कक्षापथ से पहले इसे लोड करना समाप्त हो गया।

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

था ही समस्या का सामना करना पड़ रहा है, मैं पोस्ट करेंगे क्या मेरे लिए काम किया, यह पता चला है KafkaLog4jAppender वर्ग पैकेज का नाम काफ्का 0.9 में बदल गया है, यहाँ है कि मैं क्या किया था, पोम में निर्भरता निम्नलिखित जोड़ा

<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