# 8.4 파이프라인 - Pipeline Aggregations

&#x20; Aggregation 중에는 다른 **metrics** aggregation의 결과를 새로운 입력으로 하는 **pipeline** aggregation이 있습니다. pipeline 에는 다른 버킷의 결과들을 다시 연산하는 **min\_bucket**, **max\_bucket**, **avg\_bucket**, **sum\_bucket**, **stats\_bucket**, 이동 평균을 구하는 **moving\_avg**, 미분값을 구하는 **derivative**, 값의 누적 합을 구하는 **cumulative\_sum** 등이 있습니다. Pipeline aggregation 은 `"buckets_path": "<버킷 이름>"` 옵션을 이용해서 입력 값으로 사용할 버킷을 지정합니다. 다음은 my\_stations 에서 **date\_histogram**을 이용해서 월별로 나눈 passangers 의 합계 sum을 다시 **cumulative\_sum**을 이용해서 누적값을 구하는 예제입니다.

{% tabs %}
{% tab title="request" %}
{% code title="passangers 의 값을 입력으로 받는 cumulative\_sum aggs 실행" %}

```javascript
GET my_stations/_search
{
  "size": 0,
  "aggs": {
    "months": {
      "date_histogram": {
        "field": "date",
        "interval": "month"
      },
      "aggs": {
        "sum_psg": {
          "sum": {
            "field": "passangers"
          }
        },
        "accum_sum_psg": {
          "cumulative_sum": {
            "buckets_path": "sum_psg"
          }
        }
      }
    }
  }
}
```

{% endcode %}
{% endtab %}

{% tab title="response" %}
{% code title="passangers 의 값을 입력으로 받는 cumulative\_sum aggs 실행 결과" %}

```javascript
{
  "took" : 7,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 10,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "months" : {
      "buckets" : [
        {
          "key_as_string" : "2019-06-01T00:00:00.000Z",
          "key" : 1559347200000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 7726.0
          },
          "accum_sum_psg" : {
            "value" : 7726.0
          }
        },
        {
          "key_as_string" : "2019-07-01T00:00:00.000Z",
          "key" : 1561939200000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 12699.0
          },
          "accum_sum_psg" : {
            "value" : 20425.0
          }
        },
        {
          "key_as_string" : "2019-08-01T00:00:00.000Z",
          "key" : 1564617600000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 11545.0
          },
          "accum_sum_psg" : {
            "value" : 31970.0
          }
        },
        {
          "key_as_string" : "2019-09-01T00:00:00.000Z",
          "key" : 1567296000000,
          "doc_count" : 3,
          "sum_psg" : {
            "value" : 9054.0
          },
          "accum_sum_psg" : {
            "value" : 41024.0
          }
        },
        {
          "key_as_string" : "2019-10-01T00:00:00.000Z",
          "key" : 1569888000000,
          "doc_count" : 1,
          "sum_psg" : {
            "value" : 971.0
          },
          "accum_sum_psg" : {
            "value" : 41995.0
          }
        }
      ]
    }
  }
}
```

{% endcode %}
{% endtab %}
{% endtabs %}

&#x20; 위 결과에서 **accum\_sum\_psg** 결과에 **sum\_psg** 값이 다음과 같이 계속 누적되어 더해지고 있는 것을 확인할 수 있습니다.

* `"accum_sum_psg" : { "value" : 7726.0}` = 7726.0
* `"accum_sum_psg" : { "value" : 20425.0}` = 7726.0 + 12699.0
* `"accum_sum_psg" : { "value" : 31970.0}` = 7726.0 + 12699.0 + 11545.0
* ...

&#x20; 서로 다른 버킷에 있는 값들도 bucket\_path에 `>` 기호를 이용해서 `"부모>자녀"` 형태로 지정이 가능합니다. 다음은 sum\_bucket 을 이용해서 **mon>sum\_psg** 버킷에 있는 passangers 필드값의 합을 구하는 예제입니다.

{% tabs %}
{% tab title="request" %}
{% code title="다른 부모의 자녀 버킷에 있는 필드를 입력으로 받는 pipeline aggs" %}

```javascript
GET my_stations/_search
{
  "size": 0,
  "aggs": {
    "mon": {
      "date_histogram": {
        "field": "date",
        "interval": "month"
      },
      "aggs": {
        "sum_psg": {
          "sum": {
            "field": "passangers"
          }
        }
      }
    },
    "bucket_sum_psg": {
      "sum_bucket": {
        "buckets_path": "mon>sum_psg"
      }
    }
  }
}
```

{% endcode %}
{% endtab %}

{% tab title="response" %}
{% code title="다른 부모의 자녀 버킷에 있는 필드를 입력으로 받는 pipeline aggs 실행 결과" %}

```javascript
{
  "took" : 4,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 10,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "mon" : {
      "buckets" : [
        {
          "key_as_string" : "2019-06-01T00:00:00.000Z",
          "key" : 1559347200000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 7726.0
          }
        },
        {
          "key_as_string" : "2019-07-01T00:00:00.000Z",
          "key" : 1561939200000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 12699.0
          }
        },
        {
          "key_as_string" : "2019-08-01T00:00:00.000Z",
          "key" : 1564617600000,
          "doc_count" : 2,
          "sum_psg" : {
            "value" : 11545.0
          }
        },
        {
          "key_as_string" : "2019-09-01T00:00:00.000Z",
          "key" : 1567296000000,
          "doc_count" : 3,
          "sum_psg" : {
            "value" : 9054.0
          }
        },
        {
          "key_as_string" : "2019-10-01T00:00:00.000Z",
          "key" : 1569888000000,
          "doc_count" : 1,
          "sum_psg" : {
            "value" : 971.0
          }
        }
      ]
    },
    "bucket_sum_psg" : {
      "value" : 41995.0
    }
  }
}
```

{% endcode %}
{% endtab %}
{% endtabs %}

&#x20; 이번 장에서는 Elasticsearch가 텍스트 검색엔진을 넘어 데이터 분석 엔진으로서의 기능을 가능하게 해 준 **Aggregations** 에 대해서 알아보았습니다. Aggregations 에는 다양한 값들을 연산하는 **metrics**, 범위나 종류 별로 값들을 분리하는 **bucket**, 그리고 다른 aggregation의 결과를 입력으로 받아 새로운 연산을 수행하는 **pipeline** 이 있습니다. 이 장에서는 주로 사용되는 aggregation들 위주로 기본적인 사용 방법에 대해 설명했습니다. 지금까지 설명한 것 외에도 수많은 종류의 aggregation 들이 있으니 공식 도큐먼트에서 확인 하시기 바랍니다.