Puntos clave: Esta guía detalla cómo actualizar un documento de ElasticSearch usando Python con Apache Spark. Proporciona un ejemplo sencillo y los requisitos previos necesarios, enfatizando la utiliza del conector Elasticsearch-Hadoop para Python. La guía incluye fragmentos de código disponibles en GitHub y explica conceptos clave como la recuperación de ID de documento y las operaciones de actualización.
- Código de ejemplo para actualizar documentos de ElasticSearch está disponible en GitHub.
- Los requisitos previos incluyen ElasticSearch, Apache Spark y el archivo jar Hadoop-ElasticSearch.
Aquí explicamos cómo escribir Python para programar y actualizar un documento ElasticSearch desde un Dataframe Apache Spark y RDD.
Son pocas instrucciones en internet. Las escritas por ElasticSearch son difíciles de entender y no ofrecen ejemplos. Así que aquí Haz el ejemplo más sencillo posible.
Este código añade campos adicionales a un documento JSON de ElasticSearch (ES). es decir, actualiza el documento. Spark tiene soporte nativo incorporado para Scala y Java. Pero para Python te tener que utiliza el conector Elasticsearch-Hadoop, escrito por ElasticSearch. Eso complica esta operación.
Código en Github
El código de este ejercicio está aquí:
Ejecutar código con spark-submit
Requisitos previos
- ES. Descarga el binario y no utilices apt-get instala, ya que la versión almacenada allí está demasiado antigua.
- Chispa Apache.
- Archivo jar Hadoop-ElasticSearch. Cuando te descargues desde aquí, te proporcionará jars para varios idiomas.
Añadir datos
Primero necesitamos añadir dos registros de datos a ES. La clave está la notación: school/doc/1
Significa añadir esto a la escuela del índice, escribe doc, con 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"
}'
Aquí están el código para ejecutar el código Python que aparece abajo como un trabajo de spark-submet. te no necesitas esto para avanzar por el código línea a línea con pyspark.
#!/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
Código explicado
read.format abre una conexión con ES. El elemento importante a tener en cuenta están "es.read.metadata", "true". Necesitamos los metadatos, ya que proporcionan el documento _id que necesitaremos para realizar la operación de actualización
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")
Luego leemos todos los documentos de la escuela index.
df = reader.load("school")
df.show()
Ahora filtramos para encontrar el único registro que están igual al de Harvard.
df.filter(df["school"] == "Harvard").show()
Ahora que la operación df.filter() funciona in situ. En otros no devuelve un nuevo dataframe. En su lugar, si opera sobre el dataframe actual. Te Ve eso a continuación. Solo Harvard están mostrados después de la operación de filtro.
df.show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
| Clemson|[_index -> school...|
+---------+-------+--------------------+
df.filter(df["school"] == "Harvard").show()
+---------+-------+--------------------+
| school| _metadata|
+---------+-------+--------------------+
| Harvard|[_index -> school...|
+---------+-------+--------------------+
Necesitamos recuperar el _id del _metadata, que están un Mapa. Así que convertimos esto en un RDD y luego utiliza el método collect() para Haz un objeto Fila. Después leemos la primera fila, la segunda columna por índice, y recuperamos el _id por nombre. Después utiliza df.withColumn y lit para escribir ese valor como una nueva columna con un valor constante en el dataframe df.
Nota: solo está una fila en el dataframe. Si hay más, tendríamos que realizar una operación de mapeo en el resto del código de abajo para actualizar todos los registros del dataframe.
r=df.rdd.collect()
id = r[0][1]['_id']
df2=df.withColumn("_id", lit(id))
A continuación, Crea el objeto esconf como un diccionario. En ejemplos de código en internet te Ve como JSON. Pero la notación de abajo es más limpia y fácil de manejar.
Los elementos importados a destacar son:
| es.update.script.inline | ctx._source.ubicación significa actualizar o Crea un campo llamado ubicación. ctx_source están el objeto ES para que lo haga. |
| es.update.script.params | Ubicación: son los valores de los parámetros pasados al script en línea es.update.script.inline. El <> significa escribir un literal. Si quisiéramos escribir un valor de campo, lo dejaríamos desactivado. |
| es.mapping.id | Esto le indica a ES que busque en el dataframe la columna id y utiliza eso como el ID del documento _id. ES usa eso para encontrar el documento que queremos actualizar en ES. |
| es.write.operation | upsert significa añadir el documento si no existe, de lo contrario actualizarlo. actualizar significa actualizarlo. |
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"
Esto escribe los datos. El **esconf significa leer el diccionario escconf. El modo ("append") significa añadir los campos al documento existente.
df2.write.format("org.elasticsearch.spark.sql").options(**esconf).mode("append").save("school/info")
Ahora miramos el documento y vemos que el campo de ubicación se ha actualizado a Cambridge. Un montón de snobs de la Ivy League.
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"}}]}}
Código completo
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")
Guía de ElasticSearch
- Tutorial de ElasticSearch para principiantes: Conceptos básicos de ElasticSearch
- Uso de Apache Hive con ElasticSearch
- Uso de ElasticSearch con Apache Spark
- Hoja de trucos de comandos de Curl ElasticSearch
- Ejemplo y explicación de actualización y explicación de Spark ElasticSearch Hadoop
- Cómo escribir datos de Apache Spark en ElasticSearch usando Python
- Sintaxis de búsqueda ElasticSearch y búsquedas booleanas y de agregación
- Consultas anidadas de ElasticSearch: Cómo buscar documentos incrustados
- ElasticSearch se une: Has_Child, consulta Has_parent
- Cómo configurar un clúster Elastic versión 7
- Cómo configurar Filebeat para nginx y ElasticSearch
- Uso de Beats y Logstash para enviar registros a ElasticSearch
- Cómo cargar un archivo CSV en ElasticSearch con Logstash
- Explicación de las agregaciones de ElasticSearch
- ¿Qué es el ELK Stack?
- Logstash 101: Uso de Logstash en una Canalización de Procesamiento de Datos
- Usar Kibana para ejecutar consultas en ElasticSearch usando el lenguaje de consultas Lucene y Kibana
- Aprendizaje automático ElasticSearch
- Cómo instalar Elastic Enterprise Search
- Cómo utilizar Elastic Enterprise Search con GitHub
Estas publicaciones son mías y no representan necesariamente la posición, estrategias u opinión de BMC.