Verwendung von ElasticSearch mit Apache Spark

Verwendung von ElasticSearch mit Apache Spark

  • Orchestrierung
3 Minuten Lesezeit Aktualisiert Oct 11, 2017

Auf einen Blick: ElasticSearch integriert sich nahtlos mit Apache Spark und ermöglicht so Analysen von JSON-Daten. Organisationen nutzen ElasticSearch für die Logverarbeitung, was die Abläufe und die Cybersicherheit verbessert. Mit der nativen Unterstützung von Spark für ElasticSearch seit Version 2.1 können Nutzer Connectors für fortschrittliche Datenanalysen nutzen und über grundlegende Lucene-Abfragen hinausgehen.

  • ElasticSearch wird weit verbreitet für die Log-Verarbeitung verwendet.
  • Apache Spark bietet seit Version 2.1 native Unterstützung für ElasticSearch.

ElasticSearch ist eine JSON-Datenbank, die bei Log-Verarbeitungssystemen beliebt ist. Zum Beispiel nutzen Organisationen oft ElasticSearch mit logstash oder filebeat, um Webserver-Logs, Windows-Events, Linux-Syslogs und andere Daten dorthin zu senden. Anschließend nutzen sie die Kibana-Weboberfläche, um Log-Events abzufragen. All dies ist wichtig für Cybersicherheit, Betrieb usw. Seit Spark 2.1 hat Spark nun native ElasticSearch-Unterstützung integriert, die sie Elasticsearch Hadoop nennen. Das bedeutet, dass Sie Apache Pig und Hive nutzen können, um mit JSON-Dokumenten ElasticSearch zu arbeiten. ElasticSearch Spark ist ein Connector, der vor 2.1 existierte und weiterhin unterstützt wird. Hier zeigen wir, wie man ElasticSearch Spark verwendet. Diese Connectors ermöglichen es, Analysen anhand von ElasticSearch-Daten durchzuführen. ElasticSearch selbst unterstützt nur Lucene Queries, also natürliche Sprachanfragen. Man könnte also prädiktive und klassifizierte Modelle schreiben, um Cybersicherheitsereignisse zu markieren oder andere Analysen durchzuführen, etwas, das ES nicht allein macht.

Installation ElasticSearch Spark

Zuerst brauchst du ElasticSearch. Lade eine vorgefertigte Version herunter, anstatt das Docker-Image zu verwenden. Die heruntergeladene Version hat die Authentifizierung deaktiviert, was uns einige Schritte spart.

    wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.5.2.tar.gz

Und dann brauchst du Apache Spark. Lade das hier herunter und entpacke es. Keine der beiden Programme erfordert eine Konfiguration. Und es ist nicht notwendig, Spark zu starten, um es zu verwenden. Und du musst auch Hadoop nicht starten. Du musst nur ElasticSearch starten. Führe diesen Befehl als Nicht-Root-Nutzer aus, um das zu tun:

    /usr/share/elasticsearch/elasticsearch-5.5.2/bin/elasticsearch

Dann laden Sie die ElasticSearch-Jar-Datei herunter:

    wget http://central.maven.org/maven2/org/elasticsearch/elasticsearch-spark-20_2.11/5.5.2/elasticsearch-spark-20_2.11-5.5.2.jar

Starte nun die interaktive Spark-Shell und biete die JAR-Datei als Kommandozeilenoption an:

    spark-shell --jars /home/walker/Documents/jars/elasticsearch-spark-20_2.11-5.5.2.jar

Jetzt können wir das Beispielprogramm direkt von der ElasticSearch-Website nehmen und Daten in ElasticSearch erstellen. Kopieren Sie diesen Text in die Spark-Shell.

    import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._
val sc = new SparkContext(conf)
conf.set("es.index.auto.create", "true")
val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
val airports = Map("arrival" -> "Otopeni", "SFO" -> "San Fran")
sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")

Alles, was das bewirkte, war, ein RDD zu ElasticSearch zu schreiben. Jetzt können wir ein paar einfache Zeilen Scala schreiben, um diese Daten aus ES auszulesen und sie wieder zu einem RDD zu machen.

    import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._
import org.apache.spark.SparkConf
val RDD = sc.esRDD("spark/docs")
RDD.first()

Das gibt die untenstehenden Zeilen zurück. Beachte, dass das RDD eine Map ist. Es könnte auch eine Scala Case Class oder JavaBean sein. Das sind also die Arten von Objekten, die man dort speichern kann, was Sinn macht, da das ES-Format JSON ist.

    RDD: org.apache.spark.rdd.RDD[(String, scala.collection.Map[String,AnyRef])] = ScalaEsRDD[0] at RDD at AbstractEsRDD.scala:37
res4: (String, scala.collection.Map[String,AnyRef]) = (AV5b8VamL-9DBQqpL3kV,Map(arrival -> Otopeni, SFO -> San Fran))

Jetzt können wir den ES-Index abfragen und sehen, dass der Spark-Index dort ist:

    curl 'localhost:9200/_cat/indices?v'

Antwortet:

    health status index uuid                   pri rep docs.count docs.deleted store.size pri.store.size
yellow open   spark n9aTT2HcTJ-RiRZ5Ct1Wiw   5   1          2            0      7.9kb          7.9kb

Diese Beiträge sind meine eigenen und spiegeln nicht unbedingt die Position, Strategien oder Meinung von BMC wider.

Mehr aus Orchestrierung

Interessieren Sie sich für mehr von BMC?

The Brief ist unsere monatliche E-Mail zum Mainframe und zur Enterprise-Orchestrierung: Recherchen, Kundenberichte, Produktnachrichten, Events.

Mit der Registrierung für dieses Angebot bestätigen Sie, dass Sie die Datenschutzrichtlinie von BMC gelesen und zugestimmt haben.

Sie haben The Brief abonniert

Das Briefing umfasst die Mainframe- und Enterprise-Orchestrierung – BMC AMI und Control-M – sowie die Analystenforschung, Kundeneinsätze und Events in beiden Bereichen.