Auf einen Blick: Dieser Leitfaden erklärt, wie man ein ElasticSearch-Dokument mit Python und Apache Spark aktualisiert. Er bietet ein einfaches Beispiel und notwendige Voraussetzungen, wobei der Einsatz des Elasticsearch-Hadoop-Connectors für Python betont wird. Der Leitfaden enthält Codeschnipsel, die auf GitHub verfügbar sind, und erklärt wichtige Konzepte wie Dokument-ID-Abruf und Aktualisierungsoperationen.
- Beispielcode zur Aktualisierung von ElasticSearch-Dokumenten ist auf GitHub verfügbar.
- Voraussetzungen sind ElasticSearch, Apache Spark und Hadoop-ElasticSearch Jar-Datei.
Hier erklären wir, wie man Python in Code schreibt, um ein ElasticSearch-Dokument aus einem Apache Spark Dataframe und RDD zu aktualisieren.
Im Internet gibt es nur wenige Anleitungen. Die von ElasticSearch verfassten Anleitungen sind schwer zu verstehen und bieten keine Beispiele. Deshalb machen wir hier das einfachste mögliche Beispiel.
Dieser Code fügt einem ElasticSearch (ES) JSON-Dokument zusätzliche Felder hinzu. Das heißt, er aktualisiert das Dokument. Spark hat eingebaute Unterstützung für Scala und Java. Für Python muss man jedoch den Elasticsearch-Hadoop-Connector verwenden, der von ElasticSearch geschrieben wurde. Das macht diese Operation komplizierter.
Code auf Github
Der Code für diese Übung findet sich hier:
Code mit spark-submit ausführen
Voraussetzungen
- ES. Lade die Binärdatei herunter und benutze nicht apt-get install, da die dort gespeicherte Version zu alt ist.
- Apache Spark.
- Hadoop-ElasticSearch Jar-Datei. Wenn Sie sie hier herunterladen, werden Jars für verschiedene Sprachen bereitgestellt.
Daten hinzufügen
Zuerst müssen wir zwei Datendatensätze zu ES hinzufügen. Der Schlüssel ist die Notation: schule/dokument/1
Das bedeutet, dies zur Indexschule hinzuzufügen, schreibt doc mit id = 1.
curl -XPUT --header 'Content-Type: application/json' http://localhost:9200/school/doc/1 -d '{
"school" : "Clemson"
}'
curl -XPUT --header 'Content-Type: application/json' http://localhost:9200/schools/doc/2 -d '{
"school" : "Harvard"
}'
Hier ist der Code, um den Python-Code unten als Spark-Submitt-Job auszuführen. Du brauchst diesen nicht, um den Code Zeile für Zeile mit pyspark durchzugehen.
#!/bin/bash
NUM_CORES=*
DRIVER_MEM=3g
JARS="/usr/share/spark/spark-2.3.2-bin-hadoop2.7/jars/elasticsearch-hadoop-6.4.2/dist"
export SPARK_HOME="/usr/share/spark/spark-2.3.2-bin-hadoop2.7"
export PATH=$PATH:$SPARK_HOME/bin
export CODEPATH="/home/ubuntu/Documents/esearch"
cd $CODEPATH
$SPARK_HOME/bin/spark-submit --master local[$NUM_CORES] --driver-memory
$DRIVER_MEM --jars $JARS/elasticsearch-hadoop-6.4.2.jar
$CODEPATH/updateData.py
Code erklärt
read.format öffnet eine Verbindung zu ES. Der wichtige Punkt ist "es.read.metadata", "true". Wir benötigen die Metadaten, da sie das Dokument liefern _id, dass wir die Aktualisierungsoperation durchführen müssen
reader = spark.read.format("org.elasticsearch.spark.sql").option("es.read.metadata", "true").option("es.nodes.wan.only","true").option("es.port","9200").option("es.net.ssl","false").option("es.nodes", "http://localhost")
Als Nächstes lesen wir alle Dokumente in der Index-Schule.
df = reader.load("school")
df.show()
Nun filtern wir, um den einen Datensatz zu finden, der dem von Harvard entspricht.
df.filter(df["school"] == "Harvard").show()
Jetzt, da die df.filter()- Operation in situ funktioniert. In Other gibt sie kein neues Dataframe zurück. Stattdessen arbeitet if auf dem aktuellen Dataframe. Das können Sie unten sehen. Nur Harvard wird nach der Filteroperation angezeigt.
df.show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
| Clemson|[_index -> school...|
+---------+-------+--------------------+
df.filter(df["school"] == "Harvard").show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
+---------+-------+--------------------+
Wir müssen die _id aus dem _metadata abrufen, das ist eine Map. Wir machen daraus ein RDD und verwenden die collect()- Methode, um ein Zeilenobjekt zu erstellen. Dann lesen wir die erste Zeile, die zweite Spalte nach Index und holen die _id nach Namen ab. Dann verwenden wir df.withColumn und lit , um diesen Wert als neue Spalte mit konstantem Wert in das Dataframe df zu schreiben.
Hinweis: Es gibt nur eine Zeile im Dataframe. Wenn es mehr gibt, müssten wir eine Mapping-Operation auf den Rest des untenstehenden Codes durchführen, um alle Datensätze im Dataframe zu aktualisieren.
r=df.rdd.collect()
id = r[0][1]['_id']
df2=df.withColumn("_id", lit(id))
Als Nächstes erstellen wir das esconf-Objekt als Wörterbuch. In Codebeispielen im Internet sehen Sie dies als JSON. Aber die untenstehende Notation ist übersichtlicher und leichter zu handhaben.
Die wichtigen Importpunkte sind:
| es.update.script.inline | ctx._source.location bedeutet, ein Feld namens location zu aktualisieren oder zu erstellen . ctx_source ist das ES-Objekt dafür. |
| es.update.script.params | Standort: sind die Parameterwerte, die an das Inline-Skript es.update.script.inline übergeben werden. Das <> bedeutet, ein Literal zu schreiben. Wenn wir einen Feldwert schreiben wollten, würden wir sie weglassen. |
| es.mapping.id | Das fordert ES auf, im Dataframe nach der id-Spalte zu suchen und diese als Dokument-ID _id zu verwenden. ES verwendet das, um das Dokument zu finden, das wir in ES aktualisieren wollen. |
| es.write.operation | upsert bedeutet, das Dokument hinzuzufügen, falls es nicht existiert, ansonsten es zu aktualisieren. Update bedeutet, es zu aktualisieren. |
esconf={}
esconf["es.mapping.id"] = "_id"
esconf["es.nodes"] = "localhost"
esconf["es.port"] = "9200"
esconf["es.update.script.inline"] = "ctx._source.location = params.location"
esconf["es.update.script.params"] = "location:"
esconf["es.write.operation"] = "upsert"
Dies schreibt die Daten. Der **esconf bedeutet, das Wörterbuch escconf zu lesen. Der modus ("append") bedeutet, die Felder zum bestehenden Dokument hinzuzufügen.
df2.write.format("org.elasticsearch.spark.sql").options(**esconf).mode("append").save("school/info")
Jetzt schauen wir das Dokument nach und bemerken, dass das Standortfeld nach Cambridge aktualisiert wurde. Ein Haufen Ivy-League-Snobs.
curl -X GET 'http://localhost:9chool/info/_search'
{"took":1,"timed_out":false,"_shards":{"total":5,"successful":5,"skipped":0,"failed":0},"hits":{"total":1,"max_score":1.0,"hits":[{"_index":"school","_type":"info","_id":"2","_score":1.0,"_source":{"school":"Harvard","location":"Cambridge"}}]}}
Vollständiger Code
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
from pyspark.sql.functions import lit
conf = SparkConf().setAppName("updateSchools")
sc = SparkContext(conf=conf)
sc.setLogLevel("INFO")
spark = SQLContext(sc)
reader = spark.read.format("org.elasticsearch.spark.sql").option("es.read.metadata", "true").option("es.nodes.wan.only","true").option("es.port","9200").option("es.net.ssl","false").option("es.nodes", "http://localhost")
df = reader.load("school")
df.show()
df.filter(df["school"] == "Harvard").show()
r=df.rdd.collect()
id = r[0][1]['_id']
df2=df.withColumn("_id", lit(id))
esconf={}
esconf["es.mapping.id"] = "_id"
esconf["es.nodes"] = "localhost"
esconf["es.port"] = "9200"
esconf["es.update.script.inline"] = "ctx._source.location = params.location"
esconf["es.update.script.params"] = "location:"
esconf["es.write.operation"] = "upsert"
df2.write.format("org.elasticsearch.spark.sql").options(**esconf).mode("append").save("school/info")
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.