Puntos clave: Se son proporcionan instrucciones para escribir datos de Apache Spark en ElasticSearch usando Python, simplificando el proceso para los usuarios. La guía incluye configuraciones como descargar ES-Hadoop y configurar PySpark, junto con fragmentos de código para analizar archivos de registro de Apache y guardar datos en ElasticSearch.
- Descarga ES-Hadoop para compatibilidad.
- Utiliza configuraciones específicas para guardar datos en formato ElasticSearch.
Aquí explicamos cómo escribir datos de Apache Spark en ElasticSearch (ES) usando Python. Escribiremos los datos de registro de Apache en ES.
Este tema se está complicado por todos los malos y enrevesados ejemplos que hay en internet. Pero aquí, lo Haz fácil.
Un factor que complica están que Spark ofrece soporte nativo para escribir en ElasticSearch en Scala y Java, pero no en Python. Para te necesitas descargar ES-Hadoop, que están escrito por ElasticSearch, disponible aquí.
te entonces incluye eso en el ámbito y Haz disponible para pyspark así:
pyspark --jars elasticsearch-hadoop-6.4.1.jar
Pon PySpark para utilizar Python 3 así:
export PYSPARK_PYTHON=/usr/bin/python3
La clave para entender cómo escribir en ElasticSearch están que, aunque ES están una base de datos JSON, sí tiene un requisito. Los datos deben estar en este formato:
{ "id: { the rest of your json}}
A continuación mostramos cómo Haz esa transformación.
Al final están el código completo y está disponible en línea aquí. Aquí lo explicamos en secciones:
Análisis de archivos de registro de Apache
Leemos un log de Apache en un RDD de Spark. Luego escribimos una función parse() para leer cada cadena en grupos de expresiones regulares, seleccionamos los campos que queremos y lo pasamos como diccionario:
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
En otras palabras, cuando leemos por primera vez los registros del archivo de texto en un RDD, se ve así:
['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"']
Luego utilizamos la función rdd map() para pasar cada línea a la función parse() y así obtener esto.
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'}]
Ahora, eso parece JSON, pero aún no es JSON. Vamos a utilizar json.dumps, que según la descripción técnica en la documentación de Python "serializará el obj como un flujo formateado en JSON."
También añadimos un ID. En la configuración ES que aparece a continuación indicamos a ES qué campo será el identificador único del documento: "es.mapping.id": "doc_id".
Calculamos primero un resumen SHA sobre todo el documento JSON para Crea ese ID como un número único.
Los resultados son devueltos así. Te Ve que el ID está un número SHA muy largo delante, seguido del 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"}')]
Ahora especificamos la configuración de ElasticSearch. Los elementos importantes a tener en cuenta son:
| "es.resource": 'caminante/apache' | "Walker" están el índice y "Apache" están el tipo. Todo junto a menudo se está llamado "el índice". |
| "es.mapping.id": "doc_id" | Aquí le decimos a ES qué documento utiliza como el ID del documento, lo que están igual que decir el campo _id. |
El resto de los campos son autoexplicativos.
Luego utilizamos el método saveAsNewAPIHadoopFile() para guardar el RDD en ES. No están nada de estudio ahí, ya que la sintaxis está siempre la misma para ES, así que no está necesario entender todas las partes de eso.
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))
Ahora podemos consultar ES desde la línea de comandos y mirar un documento:
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"
}
Aquí está el código completo:
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)
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.