코드를 줄이고, 재사용하고, 재활용하세요. Logstash 수집 파이프라인을 위한 모듈형 구성 요소를 생성하는 방법을 알아봅니다.

<p>오픈 소스 데이터 처리 파이프라인인 <a href="/kr/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를 시작하면 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">generator</a>인 입력을 정의합니다. Generator 입력은 Logstash를 테스트하기 위해 작성되었으며 이 예제에서는 단일 이벤트를 생성하게 됩니다.&nbsp;
</p><pre class="prettyprint">input { 
  generator { 
    lines =&gt; ["Generated line"] 
    count =&gt; 1 
  } 
}
</pre><p><strong>입력 파일: 02_in.cfg</strong>
</p><p>이 파일은 stdin에서 수신 대기하는 Logstash 입력을 정의합니다.
</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> 파이프라인이 입력 이벤트를 시뮬레이션하기 위해 generator를 실행하고 있으므로 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 표현식을 사용하며 그 후에는 알파벳 순서로 파일을 정렬합니다. 즉, glob 표현식에서 <code>03_filter.cfg</code>가 <code>02_filter.cfg</code> 앞에 오도록 <code>my-pipeline_2</code>의 정의를 변경하더라도 각 이벤트는 <code>03_filter.cfg</code>에 정의된 필터 전에 <code>02_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="/kr/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>Logstash로의 입력으로, 주식 시장 벤치마크 값을 포함하는 CSV 파일을 사용합니다. 몇 가지 예제 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>쉼표로 분리된 값은 “시간"과 증권거래소 벤치마크 값 “DAX”, “SMI”, “CAC”, “FTSE”를 나타냅니다. 이것을 예제 Logstash 파이프라인에서 입력으로 사용하기 위해 위의 줄을 “stocks.csv”라고 하는 CSV 파일로 복사해서 붙여넣습니다.
</p>
<h2>예제 Logstash 파이프라인</h2>
<p>아래에서 다음을 수행하는 Logstash 파이프라인을 보여드리겠습니다.
</p>
<ol>
	<li>CSV 파일로부터 CSV 형식의 입력으로 주식 시장 값을 읽습니다.</li>
	<li>CSV 입력의 각 줄을 JSON 문서에 매핑하고, 거기서 CSV 열은 “시간”, “DAX”, “SMI”, “CAC”, “FTSE” 등의 JSON 필드에 매핑합니다.</li>
	<li>시간 필드를 Unix 형식으로 변환합니다.</li>
	<li><a href="/guide/en/logstash/6.5/plugins-filters-clone.html">복제 필터 플러그인</a>을 사용해 각 문서의 복사본 2개를 생성합니다(이 복사본은 원본 문서 이외의 것입니다). 복제 필터는 자동으로 새로운 문서 사본 각각에 새로운 “유형" 필드를 추가하며, 그 “유형"은 복제 배열에서 부여하는 이름에 상응합니다. 우리는 유형이 “clone_for_SMI” 또는 “clone_for_FTSE”가 되도록 정의했으며, 각 복제본은 궁극적으로는 “SMI” 또는 “FTSE” 주식 시장을 위한 데이터만 포함하게 됩니다. </li>
	<li>각 복제본에 대해,</li>
	<ol>
		<li><a href="/guide/en/logstash/6.5/plugins-filters-prune.html">정리 필터 플러그인</a>을 사용해 특정 주식 시장의 허용 목록에 포함된 필드를 제외한 모든 필드를 제거합니다.</li>
		<li>복제 기능으로 추가된 “유형"에 상응하는 각 문서에 <a href="/kr/blog/logstash-metadata">메타데이터</a>를 추가합니다. 이것은 우리가 복제 기능으로 삽입된 “유형"을 제거하는 정리 기능을 사용하고 있기 때문에 필요합니다. 이 정보는 문서가 정확한 출력으로 향하도록 하기 위해 출력 단계에서 반드시 있어야 합니다.</li>
	</ol>
	<li><a href="/guide/en/logstash/6.5/plugins-outputs-elasticsearch.html">Logstash용 Elasticsearch 출력 플러그인</a>을 사용해 각 주식 시장에 대한 문서가 다른 Elasticsearch&nbsp;출력으로 향하도록 작성합니다. 출력은 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"이라고 하는 파일에 복사하세요.
</p>
<pre class="prettyprint">## 1단계
input {
  file {
    # stocks.csv 파일을 사용하려면 경로를 편집해야 합니다
    path =&gt; "${HOME}/stocks.csv"
    # 다음은 매번 Logstas가 수행할 때마다 
    # 전체 입력을 다시 읽도록 합니다(디버깅에 유용합니다).
    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개). 
  # 각 사본은 배열에서 부여된 이름에 상응하여 
  # 추가된 “유형" 필드를 자동으로 갖게 됩니다.
  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"&nbsp;파일을 읽고 프로세싱을 완료했으면, 그 결과로 생성되는 "smi_data", "ftse_data", "stocks_original"이라고 하는 3개의 인덱스를 볼 수 있습니다.
</p>
<h2>SMI 인덱스 확인</h2>
<pre class="prettyprint">GET /smi_data/_search
</pre>
<p>이것은 다음 구조를 가진 문서를 표시하게 됩니다. "smi_data"&nbsp;인덱스에는 오직 “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" 인덱스에는 오직 “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>&nbsp;다음 구조를 가진 문서를 표시하게 됩니다. "stocks_original"&nbsp;인덱스에는 필터링되지 않은 버전의 문서 원본이 나타나는 것을 눈여겨보세요.
</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="/kr/logstash">Logstash</a> 기능의 작은 부분을 시연해 보여드렸습니다. 특히, Logstash를 사용해 여러 주식 시장으로부터 데이터를 수집한 다음, 그 데이터를 처리해 고유한 출력으로 전송하는 방법을 보여주는 예제를 소개했습니다. Logstash와 <a href="/kr/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로 전송하는 경우 그리고 이렇게 자동 생성된 <code>_id</code> 값을 Elasticsearch가 삽입하는 각 문서에 사용하는 경우 다른 <code>_id</code> 값을 가진 동일한 문서가 Elasticsearch에 여러 번 저장됩니다. 이 경우 이러한 중복 문서를 찾아서 제거해야 할 수 있습니다. 따라서 이 블로그 게시물에서는 (1) Logstash를 사용하거나 (2) Python으로 작성된 사용자 정의 코드를 사용하여 Elasticsearch에서 중복 문서를 탐지하고 제거하는 방법을 다룹니다.
</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>Elasticsearch 인덱스에서 중복 문서를 탐지하고 제거하는 데 Logstash를 사용할 수 있습니다. 이 기법은 <a href="/kr/blog/logstash-lessons-handling-duplicates">Logstash를 사용한 중복 처리에 대한 이 블로그 게시물</a>에 설명되어 있으며, 이 섹션에서는 이러한 접근 방식을 적용하는 구체적인 예를 보여드리겠습니다.
</p><p>Elasticsearch 클러스터의 인덱스에서 문서를 읽은 다음, <a href="/guide/en/logstash/current/plugins-filters-fingerprint.html">지문 필터</a>를 사용하여 <code>["CAC", "FTSE", "SMI"]</code> 필드의 해시를 기반으로 각 문서에 대한 고유한 <code>_id</code> 값을 계산한 후, 마지막으로 각 문서를 동일한 Elasticsearch 클러스터의 새 인덱스에 다시 써서 중복된 문서가 동일한 <code>_id</code>에 작성되어 제거되도록 하는 간단한 Logstash 구성을 아래 예제로 작성했습니다.
</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> 횟수가 0이 안될 수 있으므로, 이론적으로 두 개의 동일하지 않은 문서가 동일한 <code>_id</code>에 매핑되어 이러한 문서 중 하나가 손실될 수 있습니다. 대부분의 실제 사례에서는 해시 충돌 가능성이 매우 낮습니다. 서로 다른 해시 함수에 대한 상세한 분석은 이 블로그의 범위를 벗어나지만, 지문 필터에 사용되는 해시 함수는 수집 성능 및 해시 충돌 횟수에 영향을 미치기 때문에 신중하게 검토해야 합니다.
</p><p>지문 필터를 사용하여 기존 인덱스에서 중복을 제거하기 위한 간단한 Logstash 구성은 다음과 같습니다.
</p><pre class="prettyprint">input {
  # Elasticsearch의 모든 문서를 읽습니다. 
  elasticsearch {
    hosts =&gt; "localhost"
    index =&gt; "stocks"
    query =&gt; '{ "sort": [ "_doc" ] }'
  }
}
# 이 필터는 2019년 2월 18일에 업데이트되었습니다.
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;-- 원래 게시 날짜 이후에 새로운 줄이 추가되었습니다.
    }
}
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.4KB인 문서가 포함되어 있다고 가정하면 인덱스에 있는 문서의 수는 1억 2천5백만 개가 됩니다. 이 경우 128비트 md5 해시를 사용할 때 중복 제거 데이터 구조를 메모리에 저장하는 데 필요한 메모리의 양은 128비트 x 1억 2천5백만 = 2GB이며, 160비트 <code>_ids</code>에는 160비트 x 1억 2천5백만 = 2.5GB의 메모리가 추가로 필요합니다. 따라서 이 알고리즘에서 모든 관련 데이터 구조를 메모리에 유지하려면 약 4.5GB의 RAM이 필요합니다. 다음 섹션에서 설명하는 접근 방식을 적용할 수 있다면 이 메모리 사용 공간을 크게 줄일 수 있습니다.
</p><h3>알고리즘 개선</h3><p>이 섹션에서는 메모리 사용량을 줄이고 새로운 중복 문서를 지속적으로 제거할 수 있도록 알고리즘을 개선한 부분을 설명합니다.
</p><p>시계열 데이터를 저장하는 경우 중복 문서가 서로 일정 시간 내에만 발생하는 것을 알고 있다면, 인덱스에 있는 문서의 하위 집합(다른 시간 창에 해당하는 각 하위 집합)에 대해 알고리즘을 반복적으로 실행하여 이 알고리즘의 메모리 사용 공간을 개선할 수 있습니다. 예를 들어 몇 년 분량의 데이터가 있는 경우 최적의 성능을 위해 <a href="/guide/en/elasticsearch/reference/current/query-filter-context.html">필터 컨텍스트</a> 내에 있는 날짜 시간 필드에 <a href="/guide/en/elasticsearch/reference/current/query-dsl-range-query.html">범위 쿼리</a>를 사용하여 한 번에 1주일씩 단계별로 데이터 세트를 살펴볼 수 있습니다. 이렇게 하려면 알고리즘이 52회(매주 1회) 실행되어야 합니다. 이 경우 최악의 메모리 사용 공간의 1/52로 줄일 수 있습니다.
</p><p>위의 예에서는 몇 주에 걸친 중복 문서를 탐지하지 못할 수 있습니다. 간격이 2시간 이상 벌어지면 중복 문서가 발생할 수 없다는 것을 알고 있다고 가정해 보겠습니다. 그렇다면 알고리즘을 실행할 때마다 이전 알고리즘 실행으로 분석된 마지막 문서 세트와 2시간씩 겹치는 문서가 포함되도록 해야 합니다. 따라서 주간 예제에서는 중복 항목이 누락되지 않도록 170시간(1주 + 2시간) 분량의 시계열 문서를 쿼리해야 합니다.
</p><p>계속해서 인덱스에서 중복 문서를 주기적으로 지우려면 최근에 받은 문서에 이 알고리즘을 실행하면 됩니다. 이때 위와 동일한 논리가 적용됩니다. 즉, 최근에 받은 문서가 약간 오래된 문서와 함께 분석에 포함되도록 하여 중복 문서가 실수로 누락되지 않도록 해야 합니다.
</p><h3>중복 문서를 탐지하는 Python 코드</h3><p>다음 코드는 문서가 동일한지 여부를 효율적으로 평가한 후 원하는 경우 삭제할 수 있는 방법을 보여줍니다. 그러나 실수로 문서가 삭제되는 것을 방지하기 위해 이 예제에서 삭제 작업을 실제로 실행하지는 않습니다. 이러한 기능은 간단하게 추가할 수 있습니다.
</p><p>Elasticsearch에서 중복 문서를 제거하는 코드는 <a href="https://github.com/alexander-marquardt/deduplicate-elasticsearch">github에서도 찾을 수 있습니다</a>.
</p><pre class="prettyprint">#!/usr/local/bin/python3
import hashlib
from elasticsearch import Elasticsearch
es = Elasticsearch(["localhost:9200"])
dict_of_duplicate_docs = {}
# 다음 줄은 문서의 중복 여부를
# 판단하는 데 사용될 필드를 정의합니다.
keys_to_include_in_hash = ["CAC", "FTSE", "SMI"]
# 현재 검색/스크롤에 의해 반환된 문서를 처리합니다.
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()
        # hashval이 새로운 것이라면
        # dict_of_duplicate_docs에 새 키를 생성하며
        # 이 키에는 빈 어레이의 값이 할당됩니다.
        # 그런 다음 _id를 즉시 어레이로 푸시합니다.
        # hashval이 이미 존재한다면
        # 새로운 _id를 기존 어레이로 푸시합니다.
        dict_of_duplicate_docs.setdefault(hashval, []).append(_id)
# 인덱스의 모든 문서를 반복하고
# dict_of_duplicate_docs 데이터 구조를 채웁니다.
def scroll_over_all_docs():
    data = es.search(index="stocks", scroll='1m',  body={"query": {"match_all": {}}})
    # 스크롤 ID를 가져옵니다.
    sid = data['_scroll_id']
    scroll_size = len(data['hits']['hits'])
    # 스크롤하기 전에 적중 결과의 현재 배치를 처리합니다.
    populate_dict_of_duplicate_docs(data['hits']['hits'])
    while scroll_size &gt; 0:
        data = es.scroll(scroll_id=sid, scroll='2m')
        # 적중 결과의 현재 배치를 처리합니다.
        populate_dict_of_duplicate_docs(data['hits']['hits'])
        # 스크롤 ID를 업데이트합니다.
        sid = data['_scroll_id']
        # 마지막 스크롤에 반환된 결과 수를 가져옵니다.
        scroll_size = len(data['hits']['hits'])
def loop_over_hashes_and_remove_duplicates():
    # 중복 해시가 있는지 확인하기 위해
    # 문서 값의 해시를 검색합니다.
    for hashval, array_of_ids in dict_of_duplicate_docs.items():
      if len(array_of_ids) &gt; 1:
        print("********** Duplicate docs hash=%s **********" % hashval)
        # 현재 hasval에 매핑된 문서를 가져옵니다.
        matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids})
        for doc in matching_docs['docs']:
            # 이 예제에서는 중복 문서만 인쇄합니다.
            # 이 코드는 인쇄하는 대신 여기서 중복을
            # 삭제하도록 간단하게 수정할 수 있습니다.
            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에서 중복 문서를 찾아 제거하는 방법

제품

회사 소개

리소스