Sortie de l'identifiant de l'application Spark dans les logs avec Log4j

Question

J'ai un fichier Log4j personnalisé pour l'application Spark. Je voudrais sortir l'identifiant de l'application Spark ainsi que d'autres attributs comme le message et la date. La structure de la chaîne JSON ressemblerait donc à ceci :

{"name":,"time":,"date":,"level":,"thread":,"message":,"app_id":}

Maintenant, cette structure ressemble à ça :

{"name":,"time":,"date":,"level":,"thread":,"message":}

Comment puis-je définir une telle disposition pour les journaux du pilote Spark ?

Mon fichier log4j ressemble à ceci :

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j='http://jakarta.apache.org/log4j/'>

    <appender name="Json" class="org.apache.log4j.ConsoleAppender">
        <layout class="org.apache.hadoop.log.Log4Json">
            <param name="ConversionLayout" value=""/>
        </layout>
    </appender>

    <root>
        <level value="INFO"/>
        <appender-ref ref="Json"/>
    </root>
</log4j:configuration>

Answer 1

Je doute que org.apache.hadoop.log.Log4Json peut être ajustée à cette fin. D'après sa javadoc et son code source, il pourrait être assez lourd.

Bien qu'il semble que vous utilisiez Log4j 1x, son API est assez flexible et nous pouvons facilement définir notre propre mise en page en étendant org.apache.log4j.Layout .

Nous aurons besoin d'une classe de cas qui sera transformée en JSON selon la structure cible :

case class LoggedMessage(name: String,
                         appId: String,
                         thread: String,
                         time: Long,
                         level: String,
                         message: String)

Et Layout pourrait être étendu comme suit. Pour accéder à la valeur de "app_id", nous utiliserons le contexte de diagnostic mappé de Log4j.

import org.apache.log4j.Layout
import org.apache.log4j.spi.LoggingEvent
import org.json4s.DefaultFormats
import org.json4s.native.Serialization.write

class JsonLoggingLayout extends Layout {
  // required by the API
  override def ignoresThrowable(): Boolean = false
  // required by the API
  override def activateOptions(): Unit = { /* nothing */ }

  override def format(event: LoggingEvent): String = {
    // we are using json4s for JSON serialization
    implicit val formats = DefaultFormats

    // retrieve app_id from Mapped Diagnostic Context
    val appId = event.getMDC("app_id") match {
      case null => "[no_app]" // logged messages outside our app
      case defined: AnyRef => defined.toString
    }
    val message = LoggedMessage("TODO",
                                appId,
                                Thread.currentThread().getName,
                                event.getTimeStamp,
                                event.getLevel.toString,
                                event.getMessage.toString)
    write(message) + "\n"
  }

}

Enfin, lorsque la session Spark est créée, nous mettons la valeur app_id dans MDC :

import org.apache.log4j.{Logger, MDC}

// create Spark session

MDC.put("app_id", session.sparkContext.applicationId)

logger.info("-------- this is info --------")
logger.warn("-------- THIS IS A WARNING --------")
logger.error("-------- !!! ERROR !!! --------")

Cela produit les journaux suivants :

{"name":"TODO","appId":"local-1550247707920","thread":"main","time":1550247708149,"level":"INFO","message":"-------- this is info --------"}
{"name":"TODO","appId":"local-1550247707920","thread":"main","time":1550247708150,"level":"WARN","message":"-------- THIS IS A WARNING --------"}
{"name":"TODO","appId":"local-1550247707920","thread":"main","time":1550247708150,"level":"ERROR","message":"-------- !!! ERROR !!! --------"}

Et, bien sûr, n'oubliez pas de faire référence à l'implémentation dans le xml de configuration de log4j :

<appender name="Json" class="org.apache.log4j.ConsoleAppender">
  <layout class="stackoverflow.q54706582.JsonLoggingLayout" />
</appender>