减少、再利用、回收（代码）。了解如何为您的 Logstash 采集管道创建模块化组件。

<p><a href="/cn/products/logstash">Logstash</a> 是一个开源的数据处理管道，可以从一个或多个输入采集事件，对事件进行转换，并将每个事件发送到一个或多个输出。某些 Logstash 实施过程可能包括多行代码，并且可能处理来自多个输入源的事件。为了让此类实施过程更加易于维护，我在这里将会演示如何基于模块化组件创建管道来提高代码的再利用率。
</p> <strong><h2>设计初衷</h2></strong><p>Logstash 经常需要对来自多个输入源的的事件应用相同的逻辑子集合。这通常会通过下面两种方法之一来实现：&nbsp;
</p><ol>
	<li>在单一管道中处理来自不同输入源的事件，这样便可以轻松地对来自所有来源的全部事件应用相同的逻辑。在此类实施过程中，不仅有相同的逻辑，通常还有大量的条件型逻辑。所以，这一方法通常会导致 Logstash 实施过程十分复杂难懂。</li>
	<li>处理来自每个唯一输入源的事件时，执行唯一的管道。这一方法要求将相同的功能复制和拷贝到每个管道中，所以很难维护代码中的相同部分。&nbsp;</li>
</ol><p>本篇博文中演示的技巧会将模块化管道组件存储在不同文件中，然后再将这些组件组合在一起以构建管道，所以该方法能够克服上述方法中的缺陷。此方法能够降低管道复杂程度，还能消除重复代码。
</p> <strong><h2>模块化管道的构建</h2></strong><p>Logstash <a href="/guide/en/logstash/current/configuration.html">配置文件</a>由输入、筛选和输出构成，这些部分均会由 Logstash 管道予以执行。&nbsp; 在更高级的设置中，通常会<a href="/guide/en/logstash/current/multiple-pipelines.html">有一个 Logstash 实例来执行多个管道</a>。默认情况下，当您在没有参数的条件下启动 Logstash 时，其会读取一个名为 <code>pipelines.yml</code> 的文件，并为指定的管道创建实例。&nbsp;
</p><p>Logstash 的输入、筛选和输出可以存储在多个文件中，用户可通过指定一个 <a href="/guide/en/logstash/current/glob-support.html">glob 表达式</a>来选择这些文件，将它们包括在一个管道中。与 glob 表达式匹配的文件将会按照字母顺序组合在一起。由于筛选的执行顺序通常十分重要，所以下面这一点会十分有帮助：在文件名称中包括一个数字标识符，以便确保文件会按照希望的顺序进行组合。
</p><p>我们在下面会定义两个特有管道，它们均由多个模块化 Logstash 组件组合而成。我们将 Logstash 组件存储在下列文件中：
</p><ul>
	<li>输入声明：<code>01_in.cfg</code>, <code>02_in.cfg</code></li>
	<li>筛选声明：<code>01_filter.cfg</code>, <code>02_filter.cfg</code>, <code>03_filter.cfg</code></li>
	<li>输出声明：<code>01_out.cfg</code></li>
</ul><p>然后，我们使用下面的 <a href="/guide/en/logstash/current/glob-support.html">glob 表达式</a>来在 <code>pipelines.yml</code> 中定义由所需组件构成的管道：&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>在上面的管道配置中，文件 <code>02_filter.cfg </code>在两个管道中都有出现，这向您展示了如何在单一文件中定义和维护两个管道中的相同代码，以及多个管道如何对相同部分予以执行。
</p> <strong><h2>测试管道</h2></strong><p>在这一部分，我们将会提供文件的具体示例，这些文件将会组合在一起成为上面 <code>pipelines.yml</code> 中所定义的唯一管道。然后我们会使用这些文件运行 Logstash，并展示生成的输出。&nbsp;
</p> <strong><h3>配置文件</h3></strong><p><strong>输入文件：01_in.cfg</strong>
</p><p>此文件定义了一个输入，此输入是一个<a href="/guide/en/logstash/current/plugins-inputs-generator.html">生成器</a>。此生成器输入旨在对 Logstash 进行测试，在这个示例中，它将会生成一个事件。&nbsp;
</p><pre class="prettyprint">input { 
  generator { 
    lines =&gt; ["Generated line"] 
    count =&gt; 1 
  } 
}
</pre><p><strong>输入文件：02_in.cfg</strong>
</p><p>此文件定义了一个 Logstash 输入，此输入会听取 stdin（标准输入）。
</p><pre class="prettyprint">input { 
  stdin {}
}
</pre><p><strong>筛选文件：01_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 01" } 
  } 
}
</pre><p><strong>筛选文件：02_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 02" } 
  } 
}
</pre><p><strong>筛选文件：03_filter.cfg</strong>
</p><pre class="prettyprint">filter { 
  mutate { 
    add_field =&gt; { "filter_name" =&gt; "Filter 03" } 
  } 
}
</pre><p><strong>输出文件：01_out.cfg</strong>
</p><pre class="prettyprint">output { 
  stdout { codec =&gt;  "rubydebug" } 
}
</pre> <strong><h3>执行管道</h3></strong><p>在无任何选项的情况下启动 Logstash，这将会执行我们在前面定义的 <code>pipelines.yml</code> 文件。使用如下代码运行 Logstash：&nbsp;
</p><pre class="prettyprint">./bin/logstash
</pre><p>由于管道 <code>my-pipeline_1</code> 正在执行生成器来模拟输入事件，我们应该可在 Logstash 初始化完毕后立即看到下列输出。从中可以看出，此管道已按照我们的预期执行了 <code>01_filter.cfg</code> 和 <code>02_filter.cfg</code> 中的内容。
</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>由于另一个名为 <code>my-pipeline_2</code> 的管道正在等待 stdin（标准输入）上的输入，所以我们看到该管道尚未处理任何事件。向 Logstash 运行所在的终端中输入一些内容，然后按下 Return 来为该管道创建一个事件。完成此操作后，您应会看到类似下面的内容：&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>从上面可以看到，<code>02_filter.cfg</code> 和 <code>03_filter.cfg</code> 中的逻辑已按照我们的预期得到了应用。&nbsp;&nbsp;
</p> <strong><h3>执行顺序</h3></strong><p>请注意 Logstash 并不会关注 glob 表达式内文件的顺序。Logstash 仅会使用 glob 表达式来确定需要包括哪些文件，然后按照字母顺序对文件进行排序。也就是说，如果我们更改 <code>my-pipeline_2 </code> 的定义，即在 glob 表达式中 <code>03_filter.cfg</code> 出现在 <code>02_filter.cfg</code> 之前，每个事件仍会先通过 <code>02_filter.cfg</code> 中的筛选，然后再通过 <code>03_filter.cfg</code> 中的筛选。
</p> <strong><h2>结论</h2></strong><p>通过使用 glob 表达式，能够允许借助模块化组件（以单独文档的形式进行保存）构成 Logstash 管道。此种方法能够提高代码的易维护性、再利用率和可读性。&nbsp;
</p><p>另外提醒一下，除了本篇博文讲述的技巧，您还应该考虑<a href="/guide/en/logstash/current/pipeline-to-pipeline.html">管道至管道通信</a>，看看其能否提高 Logstash 部署过程的模块化程度。
</p>

blog-pipe-sky-banner.jpg

blog-pipe-sky-thumb.jpg

How to create maintainable and reusable Logstash pipelines

如何创建易于维护和再利用的 Logstash 管道

了解如何确保 Elasticsearch 集群内的数据与既有关系型数据保持同步。

es-database-sync-banner.jpg

es-database-sync-thumb.jpg

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

如何使用 Logstash 和 JDBC 确保 Elasticsearch 与关系型数据库保持同步

<p><a href="/cn/logstash">Logstash</a> 是一个开源的服务器端数据处理管道，可以采集和转换数据，并将数据发送到一个或多个输出。在本篇博文中，我将会通过一个例子演示如何使用 Logstash 采集来自多个股票市场的数据，并将各个不同股票市场的相应数据发送到不同的输出。要完成这一过程，需要执行下列步骤：
</p>
<ol>
	<li>为来自股票市场输入流的每个文档创建副本。</li>
	<li>筛选每个副本，以便仅包含对给定股票市场有效的字段。</li>
	<li>向每个副本中添加元数据，以表明其包含的是哪个股票市场的数据。</li>
	<li>评估每个文档中的元数据，以便指导文档生成正确的输出。</li>
</ol>
<p>请注意在本篇博文中，我并未使用<a href="/guide/en/logstash/6.5/pipeline-to-pipeline.html">管道至管道通信（在 6.5 中推出了测试版）</a>，尽管这一功能也有可能实现此处所描述的部分功能。
</p>
<h2>示例输入文件</h2>
<p>我们将包含股票市场基准值的 CSV 文件作为 Logstash 的输入。下面给出了几条 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>逗号分隔值表示 “time”（时间）和下列股票交易基准的值：“DAX”、“SMI”、“CAC” 和 “FTSE”。将上面几行复制并粘贴到一个名为 “stocks.csv” 的 CSV 文件中，以便将其用作示例 Logstash 管道中的输入。
</p>
<h2>示例 Logstash 管道</h2>
<p>下面我们将会展示如何通过 Logstash 管道完成下列步骤：
</p>
<ol>
	<li>从 CSV 文件中读取股票市场值，并作为 CSV 格式的输入。</li>
	<li>将 CSV 输入的各行映射至一个 JSON 文档，其中 CSV 输入的各列会映射至下列 JSON 字段：“time”、“DAX”、“SMI”、“CAC” 和 “FTSE”。</li>
	<li>将时间字段转换为 Unix 格式。</li>
	<li>使用 <a href="/guide/en/logstash/6.5/plugins-filters-clone.html">clone filter plugin</a>（克隆筛选插件）为每个文档创建两个副本（在原始文档基础之上创建这些副本）。克隆筛选插件会自动为每个文档副本添加一个新的 “type”（类型）字段，其中 “type” 对应的是克隆阵列中所给的名称。我们已将类型定义为 “clone_for_SMI” 或 “clone_for_FTSE”，而且每个克隆副本最终仅会包含 “SMI” 或 “FTSE” 股票市场的数据。</li>
	<li>对于每个克隆副本：</li>
	<ol>
		<li>使用 <a href="/guide/en/logstash/6.5/plugins-filters-prune.html">prune filter plugin</a>（修剪筛选插件）删除特定股票市场白名单字段之外的所有字段。</li>
		<li>为每个文档添加与通过克隆功能所添加 “type” 对应的<a href="/cn/blog/logstash-metadata">元数据</a>。这一点很有必要，因为我们使用修剪功能删除了通过克隆功能插入的 “type”，而在输出阶段中为了指导文档生成正确的输出，这一信息为必填项。</li>
	</ol>
	<li>通过<a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">适用于 Logstash 的 Elasticsearch 输出插件</a>将各个股票市场的文档写到不同的 Elasticsearch 输出中，该输出取决于我们在第 5 步中添加的元数据字段所定义的值。在下面为了简化代码，每个 Elasticsearch 输出会向一个本地 Elasticsearch 集群中的不同索引写入内容。如果需要将多个集群作为输出，那么您可以轻松编辑每个 Elasticsearch 输出声明，以指定不同的 <a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html#plugins-outputs-elasticsearch-hosts">Elasticsearch 主机</a>。</li>
</ol>
<p>下面是一个执行上述步骤的 Logstash 管道（且在备注中添加了对应的步骤编号）。&nbsp;将这一管道复制到一个名为 "clones.conf"&nbsp;的文件中并执行：
</p>
<pre class="prettyprint">## 第 1 步
input {
  file {
    # 务必编辑路径以便使用您的 stocks.csv 文件
    path =&gt; "${HOME}/stocks.csv"
    # 下列内容将会确保 Logstash 每次执行时 
    # 都会重新读取完整输入（进行故障排查时很有用）。
    start_position =&gt; "beginning"
    sincedb_path =&gt; "/dev/null"
  }
}
## 第 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'
    }
  }
## 第 3 步  
date {
    match =&gt; ['time', 'UNIX']
  }
## 第 4 步
  # 下面一行将会为每个文档 
  # 额外创建 2 个副本（算上 
  # 原始文档的话，共有 3 个）。
  # 每个副本中都会自动添加一个 "type" 字段， 
  # 此字段与阵列中所给的名称对应。
  clone {
    clones =&gt; ['clone_for_SMI', 'clone_for_FTSE']
  }
## 第 5 步
  if [type] == 'clone_for_SMI' {
    # 删除 "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" } 
    }
  } 
}
## 第 6 步
output {
  # 发送到 stdout 的下列输出仅用于故障排查目的， 
  # 您可将其删除
  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>测试 Logstash 管道</h2>
<p>如要使用示例 CSV 数据对管道进行测试，您可以执行下列命令，可能需要对其进行编辑以确保使用的是您系统的正确路径。请注意，"config.reload.automatic" 部分为可选，但是其能允许您在无需重新启动 Logstash 的条件下重新加载 "clones.conf"：
</p>
<pre class="prettyprint">./logstash -f ./clones.conf --config.reload.automatic
</pre>
<p>一旦 Logstash 读取了 "stocks.csv" 文件并处理完毕，我们可以看到生成了三个索引，名称分别为 "smi_data"、"ftse_data" 和 "stocks_original"。
</p>
<h2>检查 SMI 索引</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
<p>这将会以下列结构显示文档。注意：在 "smi_data" 索引中仅显示了 “SMI” 数据。
</p>
<pre class="prettyprint">      {
        "_index": "smi_data",
        "_type": "doc",
        "_id": "_QRskWUBsYalOV9y9hGJ",
        "_score":1,
        "_source": {
          "SMI":1688.5    
        }
      }
</pre>
<h2>检查 FTSE 索引</h2>
<pre class="prettyprint">GET /ftse_data/_search
</pre>
<p><span style="background-color: initial;">这将</span>会以下列结构显示文档。注意：在 "ftse_data"&nbsp;索引内的文档中仅会显示 “FTSE” 字段。
</p>
<pre class="prettyprint">      {
        "_index": "ftse_data",
        "_type": "doc",
        "_id":"AgRskWUBsYalOV9y9hL0",
        "_score":1,
        "_source": {
          "FTSE":2448.2
        }
      }
</pre>
<h2>检查原始文档索引</h2>
<pre class="prettyprint">GET /stocks_originals/_search
</pre>
<p><span style="background-color: initial;">这将</span>会以下列结构显示文档。注意：在 "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>结论</h2>
<p>在本篇博文中，我针对 <a href="/cn/logstash">Logstash</a> 演示了一个小型的功能子集合。具体而言，我通过一个例子演示了如何使用 Logstash 采集来自多个股票市场的数据，对其进行处理并将这些数据发送到不同的输出。如果您正在对 Logstash 和 <a href="/cn/products">Elastic Stack</a> 进行测试并且在这一过程中遇到了问题，请随时在<a href="http://discuss.elastic.co">公共论坛上</a>寻求我们的帮助。
</p>

log-split-route-banner.jpg

log-split-route-thumb.jpg

Using Logstash to Split Data and Send it to Multiple Outputs

使用 Logstash 拆分数据并将数据发送到多个输出

了解如何使用 Logstash 或定制 Python 脚本从 Elasticsearch 中检测并移除重复文档。

<p>将数据导入 Elasticsearch 的很多系统都将利用 <a href="/guide/en/elasticsearch/reference/current/docs-index_.html#_automatic_id_generation">Elasticsearch 为新插入的文档自动生成 ID 值</a>。但是，如果数据源将同一文档多次意外发送到 Elasticsearch，并且对于 Elasticsearch 插入的每个文档都使用了这种自动生成的 <code>_id</code> 值，那么这个文档就会使用不同的 <code>_id</code> 值在 Elasticsearch 中存储多次。如果发生此情况，可能就需要查找并移除此类重复文档。因此，在本篇博文中，我们将介绍如何通过以下两种方法从 Elasticsearch 中检测并移除重复文档：(1) 使用 Logstash；(2) 使用以 Python 语言编写的定制代码。
</p><h2>示例文档结构</h2><p>就本篇博文而言，我们假设 Elasticsearch 集群中的文档具有以下结构。此结构对应的数据集包含多个表示股市交易的文档。
</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>鉴于这个示例文档结构，在本篇博文中，我们主观假设：如果多个文档的 <code>["CAC", "FTSE", "SMI"]</code> 字段具有相同的值，则它们相互重复。
</p><h2>使用 Logstash 删除重复的 Elasticsearch 文档</h2><p>Logstash 可用于从 Elasticsearch 索引中检测并移除重复文档。有关此项技术的描述，请参阅<a href="/cn/blog/logstash-lessons-handling-duplicates">介绍如何使用 Logstash 处理重复文档的博文</a>，这里我们将仅演示一个应用此方法的具体示例。
</p><p>在下面的示例中，我编写了一个简单的 Logstash 配置，从 Elasticsearch 集群上的索引中读取文档，然后使用<a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">指纹筛选器</a>根据 <code>["CAC", "FTSE", "SMI"]</code> 字段的哈希为每个文档计算一个唯一的 <code>_id</code> 值，最后将每个文档写回到同一 Elasticsearch 集群上的新索引，这样重复的文档将被写入同一 <code>_id</code>，并随之被删除。
</p><p>此外，只需稍加修改，同样的 Logstash 筛选器还可应用于写入新建索引的未来文档，以确保几乎实时地移除重复文档。为实现这一目的，可以更改以下示例中的输入部分，以接受来自实时输入源的文档，而不是从现有索引中拉取文档。
</p><p>请注意，使用定制 <code>_id</code> 值（即不是由 Elasticsearch 生成的 <code>_id</code>）将会对索引操作的<a href="/guide/en/elasticsearch/reference/master/tune-for-indexing-speed.html#_use_auto_generated_ids">写入性能造成一定影响</a>。
</p><p>另外值得注意的是，在理论上，根据所使用的哈希算法，这个方法可能会导致 <code>_id</code> 值的<a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">哈希冲突</a>数不为零，这可能会使两个不同的文档被映射到同一 <code>_id</code>，从而导致其中一个文档丢失。对于大部分实践用例，发生哈希冲突的可能性极低。对不同哈希函数的详细分析不在本博文讨论范围之内，但应仔细考虑指纹筛选器中使用的哈希函数，因其对采集性能和哈希冲突数都将有影响。
</p><p>下面介绍了一个使用指纹筛选器对现有索引进行重复文档删除的简单 Logstash 配置。
</p><pre class="prettyprint">input {
  # Read all documents from Elasticsearch 
  elasticsearch {
    hosts =&gt; "localhost"
    index =&gt; "stocks"
    query =&gt; '{ "sort": [ "_doc" ] }'
  }
}
# This filter has been updated on February 18, 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;-- New line added since original post date
    }
}
output {
    stdout { codec =&gt; dots }
    elasticsearch {
        index =&gt; "stocks_after_fingerprint"
        document_id =&gt; "%{[@metadata][generated_id]}"
    }
}
</pre><h2>用于删除 Elasticsearch 重复文档的定制 Python 脚本</h2><h3>一种节省内存的方法</h3><p>如果不使用 Logstash，则可以使用定制 Python 脚本有效地完成重复文档删除。对于这种方法，我们将计算定义为唯一标识文档的 <code>["CAC", "FTSE", "SMI"]</code> 字段的哈希。然后，我们将此哈希用作 Python 字典中的一个键，其中每个字典条目的关联值将是映射到同一哈希的文档 <code>_ids</code> 的数组。
</p><p>如果多个文档具有相同哈希，则可以删除映射到同一哈希的重复文档。或者，如果担心可能会发生<a href="https://en.wikipedia.org/wiki/Collision_(computer_science)">哈希冲突</a>，则可以检查映射到同一哈希的文档内容，以查看文档是否确实相同，确认后再删除重复文档。
</p><h3>检测算法分析</h3><p>对于一个大小为 50GB 的索引，如果假设索引包含的文档平均大小为 0.4 KB，则该索引中将有 1.25 亿个文档。在这种情况下，如果使用 128 位 md5 哈希将删除重复的数据结构存储在内存中，则所需内存量约为 128 位 x 1.25 亿 = 2GB 内存，再加上 160 位 <code>_ids</code> 还需要 160 位 x 1.25 亿 = 2.5 GB 内存。因此，这一算法将需要约 4.5GB RAM 才能将所有相关数据结构保留在内存中。此时，如果可以应用下节所讨论的方法，则可以大大减少内存占用。
</p><h3>算法增强功能</h3><p>在这部分中，我们将介绍一个算法增强功能，用于减少内存使用量并持续移除新的重复文档。
</p><p>如果您存储的是时序数据，并且知道重复文档仅在短时间内彼此重复，则可通过对索引中的文档子集重复执行此算法（其中每个子集对应一个不同的时间范围）来改进其内存占用情况。例如，如果您有一年的数据，则可以使用日期时间字段上的<a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">范围查询</a>（在<a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">筛选器上下文</a>中，用于获得最佳性能），一次一周地遍历您的数据集。这就要求该算法执行 52 次（针对每周数据集执行一次），在这种情况下，这种方法可以将极端情况下的内存占用减少 52 倍。
</p><p>在上例中，您可能会担心检测不到跨周的重复文档。让我们假设您知道重复文档的间隔不能超过 2 小时，那么，您需要确保每次执行的算法都包含与前一次执行算法分析的最后一组文档重叠 2 小时的文档。因此，在上述的每周示例中，您需要查询 170 小时（1 周 + 2 小时）的时序文档，才可确保不漏掉任何重复文档。
</p><p>如果要持续定期从索引中清除重复文档，则可以对最近收到的文档执行这个算法。与上述逻辑相同，确保分析中包含最近接收的文档，并且与稍旧文档足够重叠，以确保不会无意中漏掉重复文档。
</p><h3>用于检测重复文档的 Python 代码</h3><p>以下代码演示了如何高效评估文档，以查看它们是否相同，然后根据需要删除。但是，为了防止意外删除文档，在这个示例中，我们不会实际执行删除操作。包含此功能很容易，
</p><p>也可以<a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">在 github 中找到</a>用于从 Elasticsearch 中删除重复文档的代码。
</p><pre class="prettyprint">#!/usr/local/bin/python3
import hashlib
from elasticsearch import Elasticsearch
es = Elasticsearch(["localhost:9200"])
dict_of_duplicate_docs = {}
# The following line defines the fields that will be
# used to determine if a document is a duplicate
keys_to_include_in_hash = ["CAC", "FTSE", "SMI"]
# Process documents returned by the current search/scroll
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()
        # If the hashval is new, then we will create a new key
        # in the dict_of_duplicate_docs, which will be
        # assigned a value of an empty array.
        # We then immediately push the _id onto the array.
        # If hashval already exists, then
        # we will just push the new _id onto the existing array
        dict_of_duplicate_docs.setdefault(hashval, []).append(_id)
# Loop over all documents in the index, and populate the
# dict_of_duplicate_docs data structure.
def scroll_over_all_docs():
    data = es.search(index="stocks", scroll='1m',  body={"query": {"match_all": {}}})
    # Get the scroll ID
    sid = data['_scroll_id']
    scroll_size = len(data['hits']['hits'])
    # Before scroll, process current batch of hits
    populate_dict_of_duplicate_docs(data['hits']['hits'])
    while scroll_size &gt; 0:
        data = es.scroll(scroll_id=sid, scroll='2m')
        # Process current batch of hits
        populate_dict_of_duplicate_docs(data['hits']['hits'])
        # Update the scroll ID
        sid = data['_scroll_id']
        # Get the number of results that returned in the last scroll
        scroll_size = len(data['hits']['hits'])
def loop_over_hashes_and_remove_duplicates():
    # Search through the hash of doc values to see if any
    # duplicate hashes have been found
    for hashval, array_of_ids in dict_of_duplicate_docs.items():
      if len(array_of_ids) &gt; 1:
        print("********** Duplicate docs hash=%s **********" % hashval)
        # Get the documents that have mapped to the current hashval
        matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids})
        for doc in matching_docs['docs']:
            # In this example, we just print the duplicate docs.
            # This code could be easily modified to delete duplicates
            # here instead of printing them
            print("doc=%s\n" % doc)
def main():
    scroll_over_all_docs()
    loop_over_hashes_and_remove_duplicates()
main()
</pre><h2>结论</h2><p>在本篇博文中，我们演示了两种在 Elasticsearch 中删除重复文档的方法。第一种方法使用 Logstash 移除重复文档，第二种方法使用定制 Python 脚本查找并移除重复文档。
</p><p>如果对 Elasticsearch 重复文档删除或任何其他 Elasticsearch 相关主题有疑问，请在<a href="https://discuss.elastic.co/">讨论论坛</a>中查看各种宝贵见解和信息。
</p>

Articles by Alex Marquardt

如何创建易于维护和再利用的 Logstash 管道

如何使用 Logstash 和 JDBC 确保 Elasticsearch 与关系型数据库保持同步

使用 Logstash 拆分数据并将数据发送到多个输出

如何在 Elasticsearch 中查找并移除重复文档

产品

公司

资源