Auf einen Blick: Anleitungen zum Schreiben von Apache Spark-Daten in ElasticSearch mit Python werden bereitgestellt, was den Prozess für Nutzer vereinfacht. Der Leitfaden enthält Setups wie das Herunterladen von ES-Hadoop und die Konfiguration von PySpark sowie Codeschnipsel zum Parsen von Apache-Logdateien und zum Speichern von Daten in ElasticSearch.
- Laden Sie ES-Hadoop zur Kompatibilität herunter.
- Verwenden Sie bestimmte Konfigurationen, um Daten im ElasticSearch-Format zu speichern.
Hier erklären wir, wie man Apache Spark-Daten mit Python in ElasticSearch (ES) schreibt. Wir werden Apache-Logdaten in ES schreiben.
Dieses Thema wird kompliziert gemacht, wegen all der schlechten, verworrenen Beispiele im Internet. Aber hier machen wir es einfach.
Ein erschwerender Faktor ist, dass Spark native Unterstützung für das Schreiben in ElasticSearch in Scala und Java bietet, aber nicht in Python. Für Sie müssen Sie ES-Hadoop herunterladen, das von ElasticSearch geschrieben wurde und hier verfügbar ist.
Dann bringst du das in den Scope und stellst es Pyspark so zur Verfügung:
pyspark --jars elasticsearch-hadoop-6.4.1.jar
Stelle PySpark so ein, dass es Python 3 verwendet:
export PYSPARK_PYTHON=/usr/bin/python3
Der Schlüssel zum Verständnis des Schreibens für ElasticSearch ist, dass ES zwar eine JSON-Datenbank ist, aber eine Voraussetzung hat. Die Daten müssen in folgendem Format vorliegen:
{ "id: { the rest of your json}}
Im Folgenden zeigen wir, wie man diese Transformation vollzieht.
Unten ist der vollständige Code und er ist hier online verfügbar. Hier erklären wir ihn in Abschnitten:
Parsing von Apache-Logdateien
Wir lesen ein Apache-Log in ein Spark RDD ein. Dann schreiben wir eine Parse()- Funktion, um jede Zeichenkette in reguläre Ausdrucksgruppen einzulesen, wählen die gewünschten Felder aus und geben sie als Wörterbuch zurück:
rdd = sc.textFile("/home/ubuntu/walker/apache_logs")
regex='^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$'
p=re.compile(regex)
def parse(str):
s=p.match(str)
d = {}
d['ip']=s.group(1)
d['date']=s.group(4)
d['operation']=s.group(5)
d['uri']=s.group(6)
return d
Mit anderen Worten: Wenn wir die Textdatei-Protokolle zuerst in ein RDD einlesen, sieht es so aus:
['83.149.9.216 - - [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"']
Dann verwenden wir die rdd map() -Funktion, um jede Zeile in die Parse()- Funktion zu übertragen, um diese zu erhalten.
rdd2 = rdd.map(parse)
rdd2.take(1)
[{'date': '17/May/2015:10:05:03 +0000', 'ip': '83.149.9.216', 'operation': 'GET', 'uri': '/presentations/logstash-monitorama-2013/images/kibana-search.png'}]
Das sieht zwar nach JSON aus, aber es ist noch kein JSON. Wir werden json.dumps verwenden, das laut der technischen Beschreibung in der Python-Dokumentation "obj als JSON-formatierten Stream serialisiert."
Wir fügen außerdem eine ID hinzu. In der untenstehenden ES-Konfiguration sagen wir ES, welches Feld die eindeutige Dokumentkennung sein wird: "es.mapping.id": "doc_id".
Wir berechnen zuerst einen SHA-Digest über das gesamte JSON-Dokument, um diese ID als eindeutige Nummer zu erstellen.
Die Ergebnisse werden so zurückgegeben. Man sieht, dass die ID eine sehr lange SHA-Nummer vorne ist, gefolgt vom JSON.
rdd3.take(1)
[('a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c', '{"date": "17/May/2015:10:05:03 +0000", "ip": "83.149.9.216", "operation": "GET", "doc_id": "a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c", "uri": "/presentations/logstash-monitorama-2013/images/kibana-search.png"}')]
Nun geben wir die ElasticSearch-Konfiguration an. Die wichtigen Punkte sind:
| "es.resource" : 'Walker/Apache' | "Walker" ist der Index und "Apache" ist der Typ. Das Ganze zusammen wird oft "the index" genannt. |
| "es.mapping.id": "doc_id" | Hier sagen wir ES, welches Dokument als Dokument-ID verwendet werden soll, was dasselbe ist wie das _id Feld. |
Die übrigen Felder erklären sich von selbst.
Dann verwenden wir die saveAsNewAPIHadoopFile()- Methode, um das RDD als ES zu speichern. Dort gibt es keine Studien, da die Syntax für ES immer gleich ist, sodass man nicht alle Teile davon verstehen muss.
es_write_conf = {
"es.nodes" : "localhost",
"es.port" : "9200",
"es.resource" : 'walker/apache',
"es.input.json": "yes",
"es.mapping.id": "doc_id"
}
rdd3.saveAsNewAPIHadoopFile(
path='-',
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", keyClass="org.apache.hadoop.io.NullWritable",
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf=es_write_conf)
rdd3 = rdd2.map(addID)
def addId(data):
j=json.dumps(data).encode('ascii', 'ignore')
data['doc_id'] = hashlib.sha224(j).hexdigest()
return (data['doc_id'], json.dumps(data))
Jetzt können wir ES über die Kommandozeile abfragen und ein Dokument betrachten:
curl http://localhost:9200s/walker/apache/_search?pretty=true&?q=*
{
"_index" : "walker",
"_type" : "apache",
"_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
"_score" : 1.0,
"_source" : {
"date" : "17/May/2015:10:05:32 +0000",
"ip" : "91.177.205.119",
"operation" : "GET",
"doc_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
"uri" : "/favicon.ico"
}
Hier ist der vollständige Code:
import json
import hashlib
import re
def addId(data):
j=json.dumps(data).encode('ascii', 'ignore')
data['doc_id'] = hashlib.sha224(j).hexdigest()
return (data['doc_id'], json.dumps(data))
def parse(str):
s=p.match(str)
d = {}
d['ip']=s.group(1)
d['date']=s.group(4)
d['operation']=s.group(5)
d['uri']=s.group(6)
return d
regex='^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$'
p=re.compile(regex)
rdd = sc.textFile("/home/ubuntu/walker/apache_logs")
rdd2 = rdd.map(parse)
rdd3 = rdd2.map(addID)
es_write_conf = {
"es.nodes" : "localhost",
"es.port" : "9200",
"es.resource" : 'walker/apache',
"es.input.json": "yes",
"es.mapping.id": "doc_id"
}
rdd3.saveAsNewAPIHadoopFile(
path='-',
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", keyClass="org.apache.hadoop.io.NullWritable",
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf=es_write_conf)
ElasticSearch-Leitfaden
- ElasticSearch Tutorial für Anfänger: ElasticSearch Grundlagen
- Verwendung von Apache Hive mit ElasticSearch
- Verwendung von ElasticSearch mit Apache Spark
- Curl ElasticSearch Befehle Spickzettel
- Spark ElasticSearch Hadoop-Update und Upsert Beispiel und Erklärung
- Wie man Apache Spark-Daten mit Python in ElasticSearch schreibt
- ElasticSearch-Suchsyntax sowie boolesche und Aggregationssuchen
- ElasticSearch verschachtelte Abfragen: Wie man nach eingebetteten Dokumenten sucht
- ElasticSearch Joins: Has_Child, Has_parent Abfrage
- Wie man einen elastischen Cluster der Version 7 einrichtet
- Wie man Filebeat für nginx und ElasticSearch konfiguriert
- Beats und Logstash verwenden, um Logs an ElasticSearch zu senden
- Wie man eine CSV-Datei mit Logstash in ElasticSearch lädt
- ElasticSearch-Aggregationen erklärt
- Was ist der ELK-Stack?
- Logstash 101: Verwendung von Logstash in einer Datenverarbeitungspipeline
- Verwendung von Kibana zur Ausführung von Abfragen in ElasticSearch mit Lucene und Kibana Query Language
- ElasticSearch Maschinelles Lernen
- Wie man Elastic Enterprise Search installiert
- Wie man Elastic Enterprise Search mit GitHub verwendet
Diese Beiträge sind meine eigenen und spiegeln nicht unbedingt die Position, Strategien oder Meinung von BMC wider.