java - Apache Kafka-Log4j-Appender 不工作 Spark 集群模式

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

带有 Spark 的 kafka-log4j-appender

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

<小时/>

为Spark提供kafka-log4j-appender

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

通常，您应该能够在 fat jar 中提供 kafka-log4j-appender。我以前有过一些关于 similar problem 的经验它不起作用的地方。只是因为在集群环境中，您的类路径被 Spark 覆盖。因此，如果它也不适合您，请继续。

选项 A. 手动下载 jar:

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

您实际上需要两者，因为如果没有客户端，附加程序将无法工作。
将它们放在您触发 spark-submit 的同一台机器上。
这样做的好处是，您可以根据自己的喜好命名它们。

现在进入客户端模式

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" \

或者对于集群模式

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

并记下 jar 在 hdfs 上的 application_1569430771458_10776 文件夹中的调用方式。

现在进入客户端模式

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" \

或者对于集群模式

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