Logging applicativo per ambienti distribuiti Hadoop-based

Monica FranceschiniSolution Architecture Manager

Big Data & Analytics Competency Center Engineering

Lo scenario• Ecosistema Hadoop (Cloudera)• Applicazioni Spark (Scala) distribuite, Java, Java Webapp• Utilizzo di Log4j• Consultazione dei log centralizzata• Dedicato agli sviluppatori

• Progetto Big Data per il calcolo delle curve orarie di consumo energetico• Italia• Altre countries

• Data Science Lab di Engineering con finalità analitiche

Ambiti di applicazione

?

Dove sono i log di Spark?

• I log sono distribuiti nei worker-nodes e suddivisi per application_id:$HADOOP_HOME/logs/userlogs/application_1474886780074_XXXX

• Se yarn log aggregation attivato directory = YARN_APP_LOGS_DIR: Accesso da shell: yarn logs -applicationId <app ID>Ma dopo che l’applicazione è terminata (no real-time)!!!

??#...##@@]]]!!!!!°°°°ççççç?????????

Dove sono i log di Spark?

• Near-Real-Time• dettaglio della singola istanza di application• ricerche sui log in base a diversi criteri• severity configurabile

Requisiti

PROs:• Ready-right-out-of- the-box• Open sourceCONs• Non-Hadoop-based• Indice ElasticSearch non su HDFS

Elastic stack (ELK)

• Open source• Riutilizzabile in qualsiasi distribuzione Hadoop• Near real time• Piattaforma Big Data per i log prodotti dalle applicazioni Big Data• Configurabile dinamicamente• Customizzabile• Distribuita• Scalabile• Interrogazioni su web application

Parametri di scelta

• Log4j : sistema di logging• Zookeeper : gestione distribuita di file di configurazione e notifica di

eventi significativi per il sistema • Kafka : sistema di messagging per il trasferimento dei messaggi• Flume : processing layer per deserializzazione/mapping su solr dei

messaggi• Morphline: trasformazione dei dati per indice Solr• SolrCloud : storage dei messaggi di log• Kibana on Solr (Banana): visualizzazione dei messaggi presenti su Solr

Stack tecnologico Apache

• distributed messaging system • pub-sub model• highly scalable and redundant messaging • highly available and resilient • automatic recovery

• Producer: programma che invia messaggi a Kafka• Topic: raggruppamento di messaggi simili• Key/value pair messages: la forma dei messaggi• Consumer: programma che estrae messaggi da Kafka• Consumer Group: gruppo di consumer, per agevolare scaling e HA• Offset: identificatore logico di un messaggio all’interno di una partizione• Broker: demone responsabile dell’invio, ricezione e salvataggio dei dati• Zookeeper: sistema per coordinazione distribuita e service discovery• Partition: porzione di topic• Replica: salvataggio di una partizione di dati su più nodi per durability

Apache Kafka

• File system gerarchico e distribuito • HA e fault tolerant• Fast (in-memory reads)• Gestisce partial failures• Reliable configuration system• Gestisce load balancing

Distributed coordination service

Apache Zookeeper

“Apache Flume is a distributed, reliable, and available service for efficiently collecting, aggregating, and moving large amounts of

streaming data into the Hadoop Distributed File System (HDFS). It has a simple and flexible architecture based on streaming data flows; and is robust and fault tolerant with tunable reliability

mechanisms for failover and recovery.”

Apache Flume

Spark processes+webapps

+Java applications

Log4J

Appender Log4j Kafka

sourceKafka

channelMorphline

sinkSolr

Banana

Flume

Zookeeper(Severity level)

Processing Layer

Storage Layer

Visualization Layer

Log Agent

Architettura

Flume flessibilità nella scelta dello storage e /o visualizzazione dei messaggi.

Storage Layer

HDFS

Spark processes+webapps

+Java applications

Log4J

Appender Log4j Kafka

sourceKafka

channelMorphline

sink

Flume

Zookeeper(Severity level)

Processing Layer

Log Agent

Flessibilità della soluzione

Automaticamente vengono inseriti:• utente • hostname• thread-nameManualmente:• System id • Module id• Application Name (opzionale)• Source Id (opzionale) • Logging Level (opzionale)

Appender Log4j

• Livello di logging update (severity) gestita da Zookeeper non è necessario riavviare le applicazioni

• È possibile cambiare il livello di logging di ogni appender registrato durante l'esecuzione dell'applicazione in maniera programmatica, scartando i messaggi che hanno un livello di logging inferiore a quello impostato

• Scrive il log in formato JSON su Kafka Kafka producer• Grazie a “contesti di logging”, è possibile aggiungere informazioni in

maniera dinamica ai messaggi di logging (utilizzo di moduli condivisi all'interno della stessa applicazione ospite, da “applicazioni” diverse )

Appender Log4j- funzionalità

• 3 Kafka brokers su• 3 nodi • Fattore replica 2• 1 Kafka topic con • 3 partitions • 3 solr shards

Configurazione dello stack

Un po’ di codice…

a1.sources=kafkaa1.channels=channelKafkaa1.sinks=solrSinka1.sources.kafka.kafka.bootstrap.servers=mn01:9092,mn02:9092,mn03:9092a1.sources.kafka.type=org.apache.flume.source.kafka.KafkaSourcea1.sources.kafka.zookeeperConnect=mn01,mn02,mn03:2181/kafkaa1.sources.kafka.topic=distroLogsa1.sources.kafka.batchSize=20000a1.sources.kafka.batchDurationMillis=3000a1.sources.kafka.channels=channelKafkaa1.channels.channelKafka.type=org.apache.flume.channel.kafka.KafkaChannela1.channels.channelKafka.zookeeperConnect=mn01,mn02,mn03:2181/kafkaa1.channels.channelKafka.brokerList=mn01:9092,mn02:9092,mn03:9092a1.channels.channelKafka.parseAsFlumeEvent=truea1.sinks.solrSink.type=org.apache.flume.sink.solr.morphline.MorphlineSolrSinka1.sinks.solrSink.channel=channelKafkaa1.sinks.solrSink.batchSize=20000a1.sinks.solrSink.morphlineFile=morphlines.conf

Agente Flume

………{ extractJsonPaths { flatten: true paths: { threadId: "/threadId" level: "/level" logTime: "/logTime" systemId: "/systemId" appId: "/appId" appName: "/appName" module: "/module" action: "/action" status: "status" user: "/user" message: "/message" hostname: "/hostname" body: "/body"} } }……………….

…………..solrLocator: { collection: "distroLogs" # ZooKeeper ensemble zkHost: "mn01,mn02,mn03:2181/solr" }……..

Morphline

log4j.rootLogger=INFO,stdoutlog4j.appender.mainAppLogger=it.engineering.logging.appender.DistributedLoggerAppenderlog4j.appender.mainAppLogger.AppName=SPARK_LOGlog4j.appender.mainAppLogger.SystemId=<SYSTEM>log4j.appender.mainAppLogger.module=SPARK_REClog4j.appender.mainAppLogger.SourceId=SPARK-SCALA

Spark-Log4j.properties

Test me!!!

producer

producer

producer

partition1

partition2

partition3

consumer

Zookeeper

consumer

consumerproducer

#flume agents#processi

shard1

shard2

shard3

Solr

partition2-R

partition3-R

partition1-R

Topic

Brok

er1

Brok

er2

Brok

er3

Kafka brokers e topic

Solr Search (Hue embedded)

PROs:- molti widget grafici disponibili out-of-the-box- possibilità di creazione di molteplici dashboard- funzionalità di SSO e condivisione delle dashboard sfruttando le funzionalità di HUECONs:- Dashboard non funzionano correttamente su collection di Solr distribuite su più nodi (HUE BUG)

Visualization layer: Hue

Apache Banana (https://github.com/lucidworks/banana)• Forked from Kibana• Build come .war file• Run su web-container dei Solr server o su una macchina diversa (abilitare i CORS

sui Solr Server)

PROs:- molti widget grafici disponibili out-of the box- possibilità di creazione di molteplici dashboardCONS:- Non offre nessun tipo di autenticazione

Visualization layer: Banana

Full-text searchTime series

Detail tableFacetsDashboards

Banana dashboards

Thank you!

https://creativecommons.org/licenses/by-nc-sa/3.0/Twitter: @twittmoniqueEmail: monixfranc@gmail.comSpecial thanks to Matilde and her puppies!