Reducir, volver a usar, reciclar (código). Conoce cómo crear componentes modulares para tus pipelines de ingesta de Logstash.

<p><a href="/es/products/logstash">Logstash</a> es un pipeline de procesamiento de datos open source que ingesta eventos de una o más entradas, los transforma y después envía cada evento a una o más salidas. Algunas implementaciones de Logstash pueden tener muchas líneas de código y procesar eventos de varias fuentes de entrada. Para un mejor mantenimiento de tales implementaciones, mostraremos cómo aumentar la capacidad para volver a usar el código creando pipelines a partir de componentes modulares.
</p> <strong><h2>Motivación</h2></strong><p>Suele ser necesario para Logstash aplicar un subconjunto común de lógica a los eventos de varias fuentes de entrada. Esto se logra comúnmente en una de las dos formas siguientes:&nbsp;
</p><ol>
	<li>Procesar eventos de varias fuentes de entrada diferentes en un solo pipeline de modo que una lógica común pueda aplicarse con facilidad a todos los eventos de todas las fuentes. En estas implementaciones, además de la lógica común suele haber una cantidad importante de lógica condicional. Este enfoque por lo tanto puede resultar en implementaciones de Logstash complicadas y difíciles de entender.</li>
	<li>Ejecutar un pipeline único para procesar eventos de cada fuente única de entrada. Este enfoque requiere la duplicación y copia de la funcionalidad común en cada pipeline, lo que hace que las porciones comunes del código sean difíciles de mantener.&nbsp;</li>
</ol><p>La técnica que se presenta en este blog aborda las deficiencias de los enfoques anteriores mediante el almacenamiento de componentes modulares del pipeline en distintos archivos y la posterior construcción de pipelines combinando estos componentes. Esta técnica puede reducir la complejidad del pipeline y eliminar la duplicación de código.
</p> <strong><h2>Construcción de pipeline modular</h2></strong><p>Un <a href="/guide/en/logstash/current/configuration.html">archivo de configuración</a> de Logstash consiste en entradas, filtros y salidas que ejecuta un pipeline de Logstash.&nbsp; En configuraciones más avanzadas es habitual tener una <a href="/guide/en/logstash/current/multiple-pipelines.html">instancia de Logstash que ejecuta varios pipelines</a>. De manera predeterminada, cuando inicias Logstash sin argumentos, este leerá un archivo llamado <code>pipelines.yml</code> e instanciará los pipelines especificados.&nbsp;
</p><p>Las entradas, filtros y salidas de Logstash pueden almacenarse en varios archivos que se pueden seleccionar para incluirse en un pipeline especificando una <a href="/guide/en/logstash/current/glob-support.html">expresión glob</a>. Los archivos que coinciden con una expresión glob se combinarán en orden alfabético. Como el orden de ejecución de filtros suele ser importante, puede ser útil incluir identificadores numéricos en los nombres de archivo para garantizar que los archivos se combinen en el orden deseado.
</p><p>A continuación definiremos dos pipelines únicos que son una combinación de varios componentes de Logstash modulares. Almacenamos nuestros componentes de Logstash en los archivos siguientes:
</p><ul>
	<li>Declaraciones de entrada: <code>01_in.cfg</code>, <code>02_in.cfg</code></li>
	<li>Declaraciones de filtro: 01_filter.cfg, <code>02_filter.cfg</code>, <code>03_filter.cfg</code></li>
	<li>Declaraciones de salida:<code> 01_out.cfg</code></li>
</ul><p>Con <a href="/guide/en/logstash/current/glob-support.html">expresiones glob</a>, después definimos pipelines en <code>pipelines.yml</code> que se formarán con los componentes deseados como sigue:&nbsp;
</p><pre class="prettyprint">- pipeline.id: my-pipeline_1
  path.config: "&lt;path&gt;/{01_in,01_filter,02_filter,01_out}.cfg"
- pipeline.id: my-pipeline_2
  path.config: "&lt;path&gt;/{02_in,02_filter,03_filter,01_out}.cfg"
</pre><p>En la configuración de pipelines anterior, el archivo <code>02_filter.cfg </code>está presente en ambos pipelines, lo que demuestra cómo el código que es común a ambos pipelines puede definirse y mantenerse en un solo archivo y también ejecutarse por varios pipelines.
</p> <strong><h2>Prueba de los pipelines</h2></strong><p>En esta sección proporcionamos un ejemplo concreto de los archivos que se combinarán en los pipelines únicos definidos en <code>pipelines.yml</code> arriba. Después ejecutamos Logstash con estos archivos y presentamos la salida generada.&nbsp;
</p> <strong><h3>Archivos de configuración</h3></strong><p><strong>Archivo de entrada: 01_in.cfg</strong>
</p><p>Este archivo define una entrada que es un <a href="/guide/en/logstash/current/plugins-inputs-generator.html">generador</a>. La entrada generadora está diseñada para probar Logstash y en este caso generará un solo evento.&nbsp;
</p><pre class="prettyprint">input { 
  generator { 
    lines =&gt; ["Generated line"] 
    count =&gt; 1 
  } 
}
</pre><p><strong>Archivo de entrada: 02_in.cfg</strong>
</p><p>En este archivo se define una entrada de Logstash que escucha en stdin.
</p><pre class="prettyprint">input { 
  stdin {} 
}
</pre><p><strong>Archivo de filtro: 01_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 01" } 
  } 
}
</pre><p><strong>Archivo de filtro: 02_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 02" } 
  } 
}
</pre><p><strong>Archivo de filtro: 03_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 03" } 
  } 
}
</pre><p><strong>Archivo de salida: 01_out.cfg</strong>
</p><pre class="prettyprint">output { 
  stdout { codec =&gt;  "rubydebug" } 
}
</pre> <strong><h3>Ejecución del pipeline</h3></strong><p>Iniciar Logstash sin ninguna opción ejecutará el archivo <code>pipelines.yml</code> que definimos anteriormente. Ejecuta Logstash de la siguiente manera:&nbsp;
</p><pre class="prettyprint">./bin/logstash
</pre><p>Como el pipeline <code>my-pipeline_1</code> está ejecutando un generador para simular un evento de entrada, deberíamos ver la salida siguiente tan pronto como Logstash haya terminado de inicializarse. Esto muestra que este pipeline ejecuta los contenidos de <code>01_filter.cfg</code> y <code>02_filter.cfg</code> como se esperaba.
</p><pre class="prettyprint">{ 
       "sequence" =&gt; 0, 
           "host" =&gt; "alexandersmbp2.lan", 
        "message" =&gt; "Generated line", 
     "@timestamp" =&gt; 2020-02-05T22:10:09.495Z, 
       "@version" =&gt; "1", 
    "filter_name" =&gt; [ 
        [0] "Filter 01", 
        [1] "Filter 02" 
    ] 
}
</pre><p>Como el otro pipeline llamado <code>my-pipeline_2</code> está esperando la entrada sobre stdin, no hemos visto ningún evento procesado por ese pipeline. Escribe algo en la terminal en la que se está ejecutando Logstash y presiona Return (Regresar) para crear un evento para este pipeline. Una vez que lo hayas hecho, deberías ver algo como lo siguiente:&nbsp;
</p><pre class="prettyprint">{ 
    "filter_name" =&gt; [ 
        [0] "Filter 02", 
        [1] "Filter 03" 
    ], 
           "host" =&gt; "alexandersmbp2.lan", 
        "message" =&gt; "I’m testing my-pipeline_2", 
     "@timestamp" =&gt; 2020-02-05T22:20:43.250Z, 
       "@version" =&gt; "1" 
}
</pre><p>Podemos ver en lo anterior que la lógica de&nbsp;<code>02_filter.cfg</code> y <code>03_filter.cfg</code> se aplica como se esperaba.&nbsp;&nbsp;
</p> <strong><h3>Orden de ejecución</h3></strong><p>Ten en cuenta que Logstash no presta atención al orden de los archivos en la expresión glob. Solo usa la expresión glob para determinar los archivos que debe incluir y después los ordena alfabéticamente. Es decir, incluso si cambiáramos la definición de <code>my-pipeline_2 </code>de modo que <code>03_filter.cfg</code> aparezca en la expresión glob antes que <code>02_filter.cfg</code>, cada evento pasaría por el filtro en <code>02_filter.cfg</code> antes que por el filtro definido en <code>03_filter.cfg</code>.
</p> <strong><h2>Conclusión</h2></strong><p>Usar expresiones glob permite que los pipelines de Logstash se conformen de componentes modulares, que se almacenan como archivos individuales. Esto puede mejorar la capacidad de mantener, volver a usar y leer el código.&nbsp;
</p><p>Además de la técnica documentada en este blog, también debería tenerse en cuenta la <a href="/guide/en/logstash/current/pipeline-to-pipeline.html">comunicación de pipeline a pipeline</a> para ver si puede mejorar la modularidad de implementación de Logstash.
</p>

blog-pipe-sky-banner.jpg

blog-pipe-sky-thumb.jpg

How to create maintainable and reusable Logstash pipelines

Cómo crear pipelines de Logstash que se puedan mantener y volver a usar

Conoce cómo mantener los datos del cluster de Elasticsearch sincronizados con las bases de datos relacionales existentes.

es-database-sync-banner.jpg

es-database-sync-thumb.jpg

How to keep Elasticsearch synchronized with a relational database using Logstash and JDBC

Cómo mantener Elasticsearch sincronizado con una base de datos relacional usando Logstash y JDBC

<p><a href="/es/logstash">Logstash</a> es un pipeline de procesamiento de datos de open source del lado del servidor que ingesta datos, los transforma y luego los envía a una o más salidas. En este blog, presentaremos un ejemplo que muestra cómo usar Logstash para ingestar datos de múltiples mercados de valores y enviar los datos que corresponden a cada mercado de valores único a una salida diferente. Esto se logra siguiendo estos pasos:
</p>
<ol>
	<li>Crea copias de cada documento de un flujo de entrada del mercado de valores.</li>
	<li>Filtra cada copia para ver solo campos que sean válidos para un mercado de valores determinado.</li>
	<li>Agrega metadatos a cada copia para indicar qué datos del mercado de valores contiene.</li>
	<li>Evalúa los metadatos en cada documento para dirigir el documento hacia la salida correcta.</li>
</ol>
<p>Ten en cuenta que en este blog, no usamos la <a href="/guide/en/logstash/6.5/pipeline-to-pipeline.html">comunicación de pipeline a pipeline (beta desde la versión 6.5)</a>, con la que sería posible lograr parte de la funcionalidad descrita aquí.
</p>
<h2>Archivo de entrada de ejemplo</h2>
<p>Como entrada a Logstash, usamos un archivo CSV que contiene valores de referencia del mercado de valores. A continuación, se detallan algunos ejemplos de entradas CSV:
</p>
<pre class="prettyprint">1483230600,1628.75,1678.1,1772.8,2443.6
1483232400,1613.63,1688.5,1750.5,2460.2
1483234200,1606.51,1678.6,1718,2448.2
1483236000,1621.04,1684.1,1708.1,2470.4
</pre>
<p>Los valores separados por comas representan "tiempo" y el valor de las siguientes referencias bursátiles: "DAX", "SMI", "CAC" y "FTSE". Copia y pega las líneas anteriores en un archivo CSV llamado "stocks.cvs" para usarlo como entrada en el pipeline de Logstash de ejemplo.
</p>
<h2>Pipeline de Logstash de ejemplo</h2>
<p>A continuación, presentamos un pipeline de Logstash que realiza lo siguiente:
</p>
<ol>
	<li>Lee los valores del mercado como entrada con formato CSV desde un archivo CSV.</li>
	<li>Mapea cada fila de la entrada CSV a un documento JSON, donde las columnas CSV mapean los siguientes campos JSON: "time", "DAX","SMI", "CAC" y "FTSE".</li>
	<li>Convierte el campo de hora a formato Unix.</li>
	<li>Usa el <a href="/guide/en/logstash/6.5/plugins-filters-clone.html">plugin de filtro de clones</a> para crear dos copias de cada documento (estas copias son adicionales al documento original). El filtro de clones agrega automáticamente un nuevo campo "type" a cada nueva copia del documento, en el que "type" corresponde a los nombres dados en la matriz de clones. Hemos definido los tipos como "clone_for_SMI" o "clone_for_FTSE", y cada clon al final solo contendrá datos de los mercados de valores "SMI" o "FTSE". </li>
	<li>Para cada clon:</li>
	<ol>
		<li>Usa el <a href="/guide/en/logstash/6.5/plugins-filters-prune.html">plugin prune filter</a> para quitar todos los campos, salvo aquellos en la lista blanca del mercado de valores específico.</li>
		<li>Agrega <a href="/es/blog/logstash-metadata">metadatos</a> a cada documento que se correspondan con el "tipo" que se ha agregado con la función clonar. Esto es necesario porque usamos la función eliminar, que quita el "tipo" que se insertó con la función clonar, y esta información es necesaria en la etapa de salida para direccionar el documento a la salida correcta.</li>
	</ol>
	<li>Usa el <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">plugin de salida de Elasticsearch para Logstash</a> para escribir los documentos para cada mercado de valores de manera que vayan a una salida de Elasticsearch diferente, con la salida determinada mediante el valor definido en el campo de metadatos que agregamos en el paso&nbsp;5. Para simplificar el siguiente código, cada salida de Elasticsearch escribe a un índice único en un cluster local de Elasticsearch. Si se deben usar múltiples clusters como salidas, cada declaración de salida de Elasticsearch podrá modificarse fácilmente para especificar <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-hosts">hosts de Elasticsearch</a> únicos.</li>
</ol>
<p>A continuación, detallamos un pipeline de Logstash que ejecuta los pasos anteriores (con los números de paso correspondientes detallados como comentarios).&nbsp;Copia este pipeline en un archivo llamado "clones.conf" para ejecutarlo:
</p>
<pre class="prettyprint">## PASO 1
input {
  file {
    # asegúrate de editar la ruta para usar el archivo stocks.csv
    path =&gt; "${HOME}/stocks.csv"
    # Lo siguiente asegurará la relectura de la entrada completa 
    # cada vez que se ejecute Logstash (útil para depuración).
    start_position =&gt; "beginning"
    sincedb_path =&gt; "/dev/null"
  }
}
## PASO 2
filter {
   csv {
    columns =&gt; ["time","DAX","SMI","CAC","FTSE"]
    separator =&gt; ","
    convert =&gt; { 
      'DAX' =&gt; 'float'
      'SMI' =&gt; 'float'
      'CAC' =&gt; 'float'
      'FTSE' =&gt; 'float'
    }
  }
## PASO 3  
date {
    match =&gt; ['time', 'UNIX']
  }
## PASO 4
  # La siguiente línea creará 2 copias 
  # adicionales de cada documento (es decir, incluido 
  # el original, 3 en total). 
  # Se agregará automáticamente a cada copia un campo "type" 
  # correspondiente al nombre dado en la matriz.
  clone {
    clones =&gt; ['clone_for_SMI', 'clone_for_FTSE']
  }
## PASO 5
  if [type] == 'clone_for_SMI' {
    # Elimina todo, excepto "SMI"
    prune {
       whitelist_names =&gt; [ "SMI"]
    }
    mutate {
      add_field =&gt; { "[@metadata][type]" =&gt; "only_SMI" } 
    }
  } 
  else if [type] == 'clone_for_FTSE' {
    prune {
       whitelist_names =&gt; [ "FTSE"]
    }
    mutate {
      add_field =&gt; { "[@metadata][type]" =&gt; "only_FTSE" } 
    }
  } 
}
## PASO 6
output {
  # La siguiente salida a stdout es solo para depuración 
  # y puede eliminarse
  stdout { 
    codec =&gt;  rubydebug {
      metadata =&gt; true
    }
  }
  if [@metadata][type] == 'only_SMI' {
    elasticsearch {
      index =&gt; "smi_data"
    }
  }
  else if [@metadata][type] == 'only_FTSE' {
    elasticsearch {
      index =&gt; "ftse_data"
    }
  }
  else {
    elasticsearch {
      index =&gt; "stocks_original"
    }
  }
}
</pre>
<h2>Prueba del pipeline de Logstash</h2>
<p>Para probar este pipeline con los datos CSV de ejemplo, podrías ejecutar el siguiente comando, modificándolo para asegurarte de que estás usando rutas correctas para tu sistema. Ten en cuenta que especificar "config.reload.automatic" es opcional, pero nos permite recargar automáticamente "clones.conf" sin reiniciar Logstash:
</p>
<pre class="prettyprint">./logstash -f ./clones.conf --config.reload.automatic
</pre>
<p>Una vez que Logstash ha leído el archivo "stocks.csv" y completado el procesamiento, podemos ver los tres índices resultantes llamados "smi_data", "ftse_data" y "stocks_original".
</p>
<h2>Comprueba el índice SMI</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
<p>Esto mostrará los documentos con la siguiente estructura. Ten en cuenta que solo aparecen datos "SMI" en el índice "smi_data".
</p>
<pre class="prettyprint">      {
        "_index": "smi_data",
        "_type": "doc",
        "_id": "_QRskWUBsYalOV9y9hGJ",
        "_score": 1,
        "_source": {
          "SMI": 1688.5    
        }
      }
</pre>
<h2>Comprueba el índice FTSE</h2>
<pre class="prettyprint">GET /ftse_data/_search
</pre>
<p><span style="background-color: initial;">Esto mostrará</span> los documentos con la siguiente estructura. Ten en cuenta que solo el campo "FTSE" aparece en los documentos en el índice "ftse_data".
</p>
<pre class="prettyprint">      {
        "_index": "ftse_data",
        "_type": "doc",
        "_id": "AgRskWUBsYalOV9y9hL0",
        "_score": 1,
        "_source": {
          "FTSE": 2448.2
        }
      }
</pre>
<h2>Comprueba el índice de documentos original</h2>
<pre class="prettyprint">GET /stocks_originals/_search
</pre>
<p><span style="background-color: initial;">Esto mostrará</span> los documentos con la siguiente estructura. Ten en cuenta que la versión original sin filtro de los documentos aparece en el índice "stocks_original".
</p>
<pre class="prettyprint">      {
        "_index": "stocks_original",
        "_type": "doc",
        "_id": "-QRskWUBsYalOV9y9hFo",
        "_score": 1,
        "_source": {
          "host": "Alexanders-MBP",
          "@timestamp": "2017-01-01T00:30:00.000Z",
          "SMI": 1678.1,
          "@version": "1",
          "message": "1483230600,1628.75,1678.1,1772.8,2443.6",
          "CAC": 1772.8,
          "DAX": 1628.75,
          "time": "1483230600",
          "path": "/Users/arm/Documents/ES6.3/datasets/stocks_for_clones.csv",
          "FTSE": 2443.6
        }
      }
</pre>
<h2>Conclusión</h2>
<p>En este blog, hemos demostrado un subconjunto pequeño de las capacidades de <a href="/es/logstash">Logstash</a>. Específicamente, presentamos un ejemplo que muestra cómo usar Logstash para ingestar datos desde múltiples mercados de valores, y después procesarlos y redirigir esos datos a salidas diferentes. Si estás probando Logstash y el <a href="/es/products/">Elastic Stack</a> y tienes preguntas, no dudes en buscar ayuda en nuestros <a href="http://discuss.elastic.co">foros de debate públicos</a>.
</p>

log-split-route-banner.jpg

log-split-route-thumb.jpg

Using Logstash to Split Data and Send it to Multiple Outputs

Cómo usar Logstash para dividir datos y enviarlos a múltiples salidas

Descubre cómo detectar y eliminar documentos duplicados de Elasticsearch usando Logstash o un script personalizado de Python.

<p>Muchos sistemas que dirigen datos a Elasticsearch aprovecharán los <a href="/guide/en/elasticsearch/reference/current/docs-index_.html#_automatic_id_generation">valores&nbsp;id autogenerados de Elasticsearch</a> para los documentos recientemente insertados. Sin embargo, si la fuente de datos envía accidentalmente el mismo documento varias veces a Elasticsearch y se usan dichos valores&nbsp;<code>_id</code> autogenerados para cada documento que Elasticsearch inserta, entonces este mismo documento se almacenará varias veces en Elasticsearch con diferentes valores&nbsp;<code>_id</code>. En este caso, es posible que resulte necesario encontrar y eliminar los duplicados. Por lo tanto, en este blog abordamos cómo detectar y eliminar documentos duplicados de Elasticsearch (1)&nbsp;usando Logstash o (2)&nbsp;usando un código personalizado de Python.
</p><h2>Ejemplo de estructura del documento</h2><p>En este blog, suponemos que los documentos en el cluster de Elasticsearch tienen la estructura siguiente. Esto corresponde a un set de datos que contiene documentos que representan operaciones del mercado de valores.
</p><pre class="prettyprint">   {
      "_index": "stocks",
      "_type": "doc",
      "_id": "6fo3tmMB_ieLOlkwYclP",
      "_version": 1,
      "found": true,
      "_source": {
        "CAC": 1854.6,
        "host": "Alexanders-MBP",
        "SMI": 2061.7,
        "@timestamp": "2017-01-09T02:30:00.000Z",
        "FTSE": 2827.5,
        "DAX": 1527.06,
        "time": "1483929000",
        "message": "1483929000,1527.06,2061.7,1854.6,2827.5\r",
        "@version": "1"
      }
    }
</pre><p>Dado este ejemplo de estructura del documento, en este blog suponemos arbitrariamente que si varios documentos tienen los mismos valores en los campos <code>["CAC", "FTSE", "SMI"]</code>, son duplicados entre sí.
</p><h2>Uso de Logstash para la deduplicación de documentos de Elasticsearch</h2><p>Logstash puede usarse para detectar y eliminar documentos duplicados de un índice de Elasticsearch. Esta técnica se describe en <a href="/es/blog/logstash-lessons-handling-duplicates">este blog sobre el manejo de duplicados con Logstash</a>, y en esta sección se muestra un ejemplo concreto que aplica este enfoque.
</p><p>En el ejemplo siguiente, escribimos una configuración simple de Logstash que lee documentos de un índice en un cluster de Elasticsearch, luego usa el <a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">filtro de huellas digitales</a> para calcular un valor&nbsp;<code>_id</code> único para cada documento basado en un hash de los campos <code>["CAC", "FTSE", "SMI"]</code> y, por último, vuelve a escribir cada documento en un índice nuevo en ese mismo cluster de Elasticsearch, de modo que los documentos duplicados se escribirán con el mismo <code>_id</code> y se eliminarán.
</p><p>Además, también se podría aplicar el mismo filtro de Logstash con pequeñas modificaciones a futuros documentos escritos en el índice recientemente creado para garantizar la eliminación de los duplicados casi en tiempo real. Esto podría lograrse cambiando la sección de entrada en el ejemplo a continuación para que acepte documentos de la fuente de entrada en tiempo real, en lugar de extraer los documentos de un índice existente.
</p><p>Ten en cuenta que usar valores&nbsp;<code>_id</code> personalizados (es decir, un&nbsp;<code>_id</code> que no genera Elasticsearch) <a href="/guide/en/elasticsearch/reference/master/tune-for-indexing-speed.html#_use_auto_generated_ids">afectará el rendimiento de escritura</a> de las operaciones del índice.
</p><p>También vale la pena mencionar que según el algoritmo de hash usado, este enfoque puede teóricamente resultar en una cantidad de <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">conflictos de hashes</a> distinta de cero para el valor&nbsp;<code>_id</code>, lo que teóricamente podría resultar en la asignación de dos documentos no&nbsp;idénticos al mismo&nbsp;<code>_id</code> y provocar la pérdida de uno de estos documentos. En la mayoría de los casos prácticos, la probabilidad de un conflicto de hashes posiblemente sea muy baja. El análisis detallado de las diferentes funciones de hash excede el alcance de este blog, pero la función de hash usada en el filtro de huellas digitales debe considerarse detenidamente debido a que afectará el rendimiento de ingesta y la cantidad de conflictos de hashes.
</p><p>A continuación, se proporciona una configuración simple de Logstash para deduplicar un índice existente usando el filtro de huellas digitales.
</p><pre class="prettyprint">input {
  # Leer todos los documentos de Elasticsearch 
  elasticsearch {
    hosts =&gt; "localhost"
    index =&gt; "stocks"
    query =&gt; '{ "sort": [ "_doc" ] }'
  }
}
# Filtro actualizado el 18&nbsp;de&nbsp;febrero de&nbsp;2019
filter {
    fingerprint {
        key =&gt; "1234ABCD"
        method =&gt; "SHA256"
        source =&gt; ["CAC", "FTSE", "SMI"]
        target =&gt; "[@metadata][generated_id]"
    &nbsp;&nbsp;&nbsp;&nbsp;concatenate_sources =&gt; true # &lt;-- Línea nueva agregada posteriormente a la fecha de publicación original
    }
}
output {
    stdout { codec =&gt; dots }
    elasticsearch {
        index =&gt; "stocks_after_fingerprint"
        document_id =&gt; "%{[@metadata][generated_id]}"
    }
}
</pre><h2>Script personalizado de Python para la deduplicación de documentos de Elasticsearch</h2><h3>Un enfoque eficiente desde el punto de vista de la memoria</h3><p>Si no se usa Logstash, la deduplicación puede lograrse de forma eficiente con un script personalizado de Python. Para este enfoque, calculamos el hash de los campos <code>["CAC", "FTSE", "SMI"]</code> que definimos para identificar de manera inequívoca un documento. Luego usamos este hash como una clave en un diccionario de Python, en el cual el valor asociado de cada entrada del diccionario será una matriz de las&nbsp;<code>_ids</code> de los documentos mapeados al mismo hash.
</p><p>Si más de un documento tiene el mismo hash, los documentos duplicados mapeados al mismo hash pueden eliminarse. Como alternativa, si te preocupa la posibilidad de <a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">conflictos de hashes</a>, es posible examinar los contenidos de los documentos mapeados al mismo hash para comprobar que realmente sean idénticos y, si es así, pueden eliminarse los duplicados.
</p><h3>Análisis del algoritmo de detección</h3><p>Si suponemos que un índice de 50&nbsp;GB contiene documentos de un tamaño promedio de 0.4&nbsp;kB, habría 125&nbsp;millones de documentos en el índice. En este caso, la cantidad de memoria necesaria para almacenar las estructuras de datos de deduplicación en la memoria al usar un hash&nbsp;md5 de 128&nbsp;bits sería de aproximadamente 128&nbsp;bits&nbsp;x&nbsp;125&nbsp;millones&nbsp;=&nbsp;2&nbsp;GB, y los <code>_ids</code> de 160&nbsp;bits requerirán otros 160&nbsp;bits&nbsp;x&nbsp;125&nbsp;millones&nbsp;=&nbsp;2.5&nbsp;GB de memoria. Por lo tanto, este algoritmo requerirá aproximadamente 4.5&nbsp;GB de RAM para mantener todas las estructuras de datos relevantes en la memoria. Es posible reducir drásticamente este consumo de memoria si puede aplicarse el enfoque que se analiza en la sección siguiente.
</p><h3>Mejora del algoritmo</h3><p>En esta sección, presentamos una mejora del algoritmo para reducir el uso de la memoria y eliminar de forma continua nuevos documentos duplicados.
</p><p>Si almacenas datos temporales y sabes que los documentos duplicados solo se producirán con poca diferencia de tiempo entre sí, entonces podrás mejorar el consumo de memoria de este algoritmo ejecutándolo reiteradamente en un subconjunto de los documentos del índice, donde cada subconjunto corresponda a una ventana de tiempo diferente. Por ejemplo, si cuentas con datos de todo un año, podrías usar <a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">búsquedas de rango</a> en el campo datetime (dentro de un <a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">contexto de filtro</a> para un mejor rendimiento) para avanzar por el conjunto de datos una semana a la vez. Esto requeriría ejecutar el algoritmo 52&nbsp;veces (una por semana); y en este caso, este enfoque reduciría por un factor de&nbsp;52 el consumo de memoria correspondiente al peor de los casos.
</p><p>Es posible que en el ejemplo anterior te preocupe no detectar documentos duplicados que se extienden entre semanas. Supongamos que sabes que no pueden producirse documentos duplicados con más de 2&nbsp;horas de diferencia. Entonces necesitarías asegurarte de que cada ejecución del algoritmo incluya documentos que se superponen por el lapso de 2&nbsp;horas con el último conjunto de documentos analizado en la ejecución anterior del algoritmo. En el caso del ejemplo semanal, necesitarías buscar documentos temporales por 170&nbsp;horas (1&nbsp;semana&nbsp;+&nbsp;2&nbsp;horas) para garantizar que no se pase por alto ningún duplicado.
</p><p>Si deseas eliminar periódicamente los documentos duplicados de tus índices de forma continua, puedes ejecutar este algoritmo en los documentos recibidos recientemente. Aplica la misma lógica anterior: asegúrate de incluir los documentos recibidos recientemente en el análisis y de superponer ligeramente los documentos anteriores para garantizar que no se hayan pasado por alto duplicados inadvertidamente.
</p><h3>Código de Python para detectar documentos duplicados</h3><p>El código siguiente muestra cómo se pueden evaluar los documentos de forma eficiente para comprobar si son idénticos y posteriormente eliminarlos, si así lo deseas. Sin embargo, para prevenir eliminaciones accidentales de documentos, en este ejemplo no ejecutamos realmente ninguna operación de eliminación. Sería sencillo incluir dicha funcionalidad.
</p><p>El código para deduplicar documentos de Elasticsearch también puede <a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">encontrarse en GitHub</a>.
</p><pre class="prettyprint">#!/usr/local/bin/python3
import hashlib
from elasticsearch import Elasticsearch
es = Elasticsearch(["localhost:9200"])
dict_of_duplicate_docs = {}
# La línea siguiente define los campos que
# se usarán para determinar si un documento es un duplicado.
keys_to_include_in_hash = ["CAC", "FTSE", "SMI"]
# Procesar documentos que devuelve search/scroll actual
def populate_dict_of_duplicate_docs(hits):
    for item in hits:
        combined_key = ""
        for mykey in keys_to_include_in_hash:
            combined_key += str(item['_source'][mykey])
        _id = item["_id"]
        hashval = hashlib.md5(combined_key.encode('utf-8')).digest()
        # Si el hashval es nuevo, crearemos una clave nueva
        # en dict_of_duplicate_docs, a la que se
        # asignará un valor de una matriz vacía.
        # Luego insertamos inmediatamente el&nbsp;_id en la matriz.
        # Si el hashval ya existe, entonces
        # solo insertaremos el&nbsp;_id nuevo en la matriz existente.
        dict_of_duplicate_docs.setdefault(hashval, []).append(_id)
# Realiza el bucle por todos los documentos del índice y completa
# la estructura de datos dict_of_duplicate_docs.
def scroll_over_all_docs():
    data = es.search(index="stocks", scroll='1m',  body={"query": {"match_all": {}}})
    # Obtén la ID de scroll.
    sid = data['_scroll_id']
    scroll_size = len(data['hits']['hits'])
    # Antes del scroll, procesa el batch actual de resultados.
    populate_dict_of_duplicate_docs(data['hits']['hits'])
    while scroll_size &gt; 0:
        data = es.scroll(scroll_id=sid, scroll='2m')
        # Procesa el batch actual de resultados.
        populate_dict_of_duplicate_docs(data['hits']['hits'])
        # Actualiza la ID de scroll.
        sid = data['_scroll_id']
        # Obtén la cantidad de resultados del último scroll.
        scroll_size = len(data['hits']['hits'])
def loop_over_hashes_and_remove_duplicates():
    # Busca en el hash de valores de documentos para comprobar
    # si se encontraron hashes duplicados.
    for hashval, array_of_ids in dict_of_duplicate_docs.items():
      if len(array_of_ids) &gt; 1:
        print("********** Duplicate docs hash=%s **********" % hashval)
        # Obtén los documentos mapeados al hashval actual.
        matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids})
        for doc in matching_docs['docs']:
            # En este ejemplo, simplemente imprimimos los documentos duplicados.
            # Este código podría modificarse fácilmente para eliminar los duplicados
            # aquí en lugar de imprimirlos.
            print("doc=%s\n" % doc)
def main():
    scroll_over_all_docs()
    loop_over_hashes_and_remove_duplicates()
main()
</pre><h2>Conclusión</h2><p>En este blog, mostramos dos métodos para la deduplicación de documentos en Elasticsearch. El primer método usa Logstash para eliminar los documentos duplicados, y el segundo usa un script personalizado de Python para encontrar y eliminar los documentos duplicados.
</p><p>Si tienes alguna duda sobre la deduplicación de documentos de Elasticsearch o cualquier otro tema relacionado con Elasticsearch, echa un vistazo a nuestros <a href="https://discuss.elastic.co/">foros de debate</a> para acceder a conocimientos e información valiosos.
</p>

Articles by Alex Marquardt

Cómo crear pipelines de Logstash que se puedan mantener y volver a usar

Cómo mantener Elasticsearch sincronizado con una base de datos relacional usando Logstash y JDBC

Cómo usar Logstash para dividir datos y enviarlos a múltiples salidas

Cómo encontrar y eliminar documentos duplicados en Elasticsearch

Productos

Compañía

Recursos