Apache Kafka-Log4j-Appender 不工作火花集群模式

Question

我尝试将我的 spark 应用程序驱动程序/执行程序日志发送到 kafka 主题。它在 Eclipse 中工作。
部署集群模式时显示

ClassNotFoundException。

我添加了 spark core、spark-sql、spark stream kafka、spark streaming、kafka streaming 和 kafka-log4j-appender maven，但它仍然显示

org.apche.kafka.log4jappender ClassNotFoundException。

在服务器中，我的 spark 程序运行 jar 中没有可用的与 kafka 相关的 jar。

请任何人告诉为什么它没有在集群模式下从 spark 驱动程序/执行程序向 kafka 主题发送日志。

我也尝试了以下链接提到的步骤，

spark streaming application and kafka log4j appender issue

og4j.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-M` `M-dd` ` HH:mm:ss} %-5p %c{1}:%L - %m%n` `

如何使用 log4j.properties 通过 kafka-log4j-appender-2.3.0 将 spark 驱动程序/执行程序日志发送到 kafka 主题。

Answer 1

kafka-log4j-appender 与 Spark

我设法在cluster 模式下将spark-submit 2.1.1 与kafka-log4j-appender 2.3.0 一起使用，但我相信其他版本的行为也会类似。

---

为 Spark 提供 kafka-log4j-appender

基本上，spark 缺少 kafka-log4j-appender。

一般来说，您应该能够在您的胖罐中提供kafka-log4j-appender。我以前对similar problem 有一些经验，但它不起作用。仅仅是因为在集群环境中，您的类路径被 Spark 覆盖。因此，如果它对您也不起作用，请继续。

选项A.手动下载jar：

kafka-log4j-appender-2.3.0.jar
kafka-clients-2.3.0.jar

实际上两者都需要，因为没有客户端，appender 将无法工作。
将它们放在您使用spark-submit 发射的同一台机器上。
好处是，您可以随意命名它们。

现在为client 模式

JARS='/absolute/path/kafka-log4j-appender-2.3.0.jar,/absolute/path/kafka-clients-2.3.0.jar'
JARS_CLP='/absolute/path/kafka-log4j-appender-2.3.0.jar:/absolute/path/kafka-clients-2.3.0.jar'
JARS_NAMES='kafka-log4j-appender-2.3.0.jar:kafka-clients-2.3.0.jar'

spark-submit \
    --deploy-mode client \
    --jars "$JARS"
    --conf "spark.driver.extraClassPath=$JARS_CLP" \
    --conf "spark.executor.extraClassPath=$JARS_NAMES" \

或者对于cluster模式

spark-submit \
    --deploy-mode cluster \
    --jars "$JARS"
    --conf "spark.driver.extraClassPath=$JARS_NAMES" \
    --conf "spark.executor.extraClassPath=$JARS_NAMES" \

选项B。使用--packages从maven下载jar：

我觉得这样比较方便，但是你得准确地取名字。

你需要在运行过程中寻找那些类型的行：

19/11/15 19:44:08 INFO yarn.Client: Uploading resource file:/srv/cortb/home/atais/.ivy2/jars/org.apache.kafka_kafka-log4j-appender-2.3.0.jar -> hdfs:///user/atais/.sparkStaging/application_1569430771458_10776/org.apache.kafka_kafka-log4j-appender-2.3.0.jar
19/11/15 19:44:08 INFO yarn.Client: Uploading resource file:/srv/cortb/home/atais/.ivy2/jars/org.apache.kafka_kafka-clients-2.3.0.jar -> hdfs:///user/atais/.sparkStaging/application_1569430771458_10776/org.apache.kafka_kafka-clients-2.3.0.jar

并记下在hdfs 上的application_1569430771458_10776 文件夹中如何调用jars。

现在为client 模式

JARS_CLP='/srv/cortb/home/atais/.ivy2/jars/org.apache.kafka_kafka-log4j-appender-2.3.0.jar:/srv/cortb/home/atais/.ivy2/jars/org.apache.kafka_kafka-clients-2.3.0.jar'
KAFKA_JARS='org.apache.kafka_kafka-log4j-appender-2.3.0.jar:org.apache.kafka_kafka-clients-2.3.0.jar'

spark-submit \
    --deploy-mode client \
    --packages "org.apache.kafka:kafka-log4j-appender:2.3.0"
    --conf "spark.driver.extraClassPath=$JARS_CLP" \
    --conf "spark.executor.extraClassPath=$KAFKA_JARS" \

或者对于cluster模式

spark-submit \
    --deploy-mode cluster \
    --packages "org.apache.kafka:kafka-log4j-appender:2.3.0"
    --conf "spark.driver.extraClassPath=$KAFKA_JARS" \
    --conf "spark.executor.extraClassPath=$KAFKA_JARS" \

---

上面应该已经可以工作了

额外步骤

如果您想提供您的 logging.proprietes，请在此处按照我的教程进行操作：https://stackoverflow.com/a/55596389/1549135