apache spark

您是否曾經對運行了幾個小時的Spark作業感到沮喪，但由于基礎設施問題而失敗了。

您會很晚才知道此故障，并浪費了數小時的時間，當Spark UI日志也無法進行事后檢查時，它會更加痛苦。

你不是一個人！

在本文中，我將介紹如何啟用自己的自定義記錄器，使其與Spark記錄器配合使用。

該自定義記錄器將收集從被動監視到主動監視所需的所有信息。

無需為此設置額外的日志記錄。

Spark 2.X基于Slf4j抽象，并且使用了logback綁定。

讓我們從基本的日志記錄開始，即如何在Spark作業或應用程序中獲取記錄器實例。

val _LOG = LoggerFactory.getLogger(this.getClass.getName)

就是這么簡單，現在您的應用程序使用的是與Spark基于相同的日志庫和設置。

現在要做一些更有意義的事情，我們必須注入自定義記錄器，該記錄器將收集信息并將其寫入彈性搜索或發布到某些REST端點或發送警報。

讓我們一步一步去做

構建自定義日志附加程序

由于spark 2.X是基于logback的，因此我們必須編寫logback logger。

自定義登錄記錄器的代碼段

class MetricsLogbackAppender extends UnsynchronizedAppenderBase[ILoggingEvent] {override def append(e: ILoggingEvent) = {//Send this message to elastic search or REST end pointmessageCount.compute(Thread.currentThread().getName, mergeValue)System.out.println(messageCount + " " + e)}val messageCount = new ConcurrentHashMap[String, AtomicInteger]()val mergeValue = new BiFunction[String, AtomicInteger, AtomicInteger] {def apply(key: String, currentValue: AtomicInteger) = {val nextValue = currentValue match {case null => new AtomicInteger(0)case _ => currentValue}nextValue.incrementAndGet()nextValue}}}

這是一個非常簡單的記錄器，它在每個線程中計數消息，您所要做的就是覆蓋附加函數。

這種類型的記錄器可以執行任何操作，例如寫入數據庫或發送到REST端點或發出警報。

啟用記錄器

要使用新的記錄器，請創建logback.xml文件并為新的記錄器添加條目。

該文件可以打包在Shaded jar中，也可以指定為運行時參數。

樣本logback.xml

<configuration><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><!-- encoders are assigned the typech.qos.logback.classic.encoder.PatternLayoutEncoder by default --><encoder><pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern></encoder></appender><appender name="METRICS" class="micro.logback.MetricsLogbackAppender"/><root level="info"><appender-ref ref="STDOUT" /></root><logger level="info" name="micro" additivity="true"><appender-ref ref="METRICS" /></logger><logger level="info" name="org.apache.spark.scheduler.DAGScheduler" additivity="true"><appender-ref ref="METRICS" /></logger></configuration>

此配置文件將MetricsLogbackAppender添加為METRICS

<appender name="METRICS" class="micro.logback.MetricsLogbackAppender"/>

接下來為應使用此功能的包/類啟用它

<logger level="info" name="micro" additivity="true"> <appender-ref ref="METRICS" /></logger> <logger level="info" name="org.apache.spark.scheduler.DAGScheduler" additivity="true"> <appender-ref ref="METRICS" /></logger

大功告成！

從'micro'包或DAGScheduler類記錄的任何消息都將使用new logger。

使用這種技術，執行者日志也可以被捕獲，當Spark作業在成百上千的執行者上運行時，這非常有用。

現在，它提供了很多讓BI實時顯示所有這些消息的選項，允許團隊提出問題或在情況不佳時訂閱變更。

警告：請確保此新記錄器減慢了應用程序的執行速度，建議使其異步。

在正確的時間獲取見解并將其付諸實踐

此博客中使用的代碼可在github中的@ sparkmicroservices回購中獲得。

我有興趣知道您正在為Spark使用哪種日志記錄模式。

翻譯自: https://www.javacodegeeks.com/2018/05/custom-logs-in-apache-spark.html

apache spark

總結

以上是生活随笔為你收集整理的apache spark_Apache Spark中的自定义日志的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。