16. 파이프라인 집계(Pipeline Aggregations)

  파이프라인 집계(Pipeline Aggregations)는 다른 집계와 달리 쿼리 조건에 부합하는 문서에 대해 집계를 수행하는 것이 아니라 다른 집계로 생성된 버킷을 참조해서 집계를 수행합니다. 부모(Parent), 형제(Sibling)라는 두 가지 유형이 있으며 집계를 수행할 때는 buckets_path 파라미터를 사용해 참조할 집계의 경로를 지정함으로써 체인 형식으로 집계 간의 연산이 이루어 집니다.

# 집계 참조 방법
AGG_SEPARATOR = '>';
METRIC_SEPARATOR = '.';
AGG_NAME = <집계 이름>;
METRIC = <메트릭 집계 이름(다중 메트릭 집계인 경우)>;
PATH = <AGG_NAME> [ <AGG_SEPARATOR>, <AGG_NAME> ]* [ <METRIC_SEPARATOR>, <METRIC> ];

 

1. 형제 집계(Sibling Aggregation)

  동일 선상의 위치에서 수행되는 새 집계를 의미합니다. 즉, 기존 버킷에 추가되는 형태가 아니라 동일 선상의 위치에 새 집계가 생성되는 파이프라인 집계입니다.

# 아파치 웹 로그에서 분 단위로 합산된 데이터량 중 가장 큰 데이터량을 구하는 예제
GET /apache-web-log/_search?size=0
{
  "aggs": {
    "histo": {
      "date_histogram": { # 분 단위 집계
        "field": "timestamp",
        "interval": "minute"
      },
      "aggs": {           # 분 단위 데이터량 합 집계
        "bytes_sum": {
          "sum": {
            "field": "bytes"
          }
        }
      }
    },
    "max_bytes": {                        # 파이프라인 집계 이름
      "max_bucket": {                     # 최대 버킷 집계
        "buckets_path": "histo>bytes_sum" # 참조할 버킷 집계
      }
    }
  }
}

  sum 집계는 결과가 하나만 존재하는 단일 메트릭 집계이기 때문에 집계 이름만으로 참조할 수 있지만 stats 같은 다중 메트릭 집계의 경우에는 메트릭명까지 지정해줘야 합니다. 예) histo>bytes_stats.avg

// 최대 버킷 집계
{
  "max_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 최소 버킷 집계
{
  "min_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 평균 버킷 집계
{
  "avg_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 통계 버킷 집계
{
  "stats_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 확장 통계 버킷 집계
{
  "extended_stats_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 백분위수 버킷 집계
{
  "percentiles_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

// 이동 평균 버킷 집계
{
  "moving_avg_bucket": {
    "buckets_path": "histo>bytes_sum"
  }
}

 

2. 부모 집계(Parent Aggregation)

  집계를 통해 생성된 버킷을 사용해 계산을 수행하고, 그 결과를 기존 집계에 반영합니다.

GET /apache-web-log/_search?size=0
{
  "aggs": {
    "histo": { # 부모 히스토그램. 일 단위 집계 수행
      "date_histogram": {
        "field": "timestamp",
        "interval": "day"
      },
      "aggs": { # 일 단위 데이터 크기를 합산하는 sum 집계 수행
        "bytes_sum": {
          "sum": {
            "field": "bytes"
          }
        },
        "sum_deriv": { # 일 단위 데이터 크기의 합산된 값을 버킷의 현재 값과 이전 값을 비교하는 집계 수행
          "derivative": {
            "buckets_path": "bytes_sum"
          }
        }
      }
    }
  }
}

  파생 집계의 경우 선행되는 데이터가 존재하지 않으면 집계를 수행할 수 없기 때문에 min_doc_count 값을 0으로 설정해야 합니다. 데이터가 존재하지 않는 부분을 갭(gap)이라 하며 파이프라인 집계에 원하는 동작을 알리는 메커니즘으로 갭 정책(gap_policy)가 있습니다. 모든 파이프라인 집계에서는 gap_policy 파라미터를 허용합니다.

skip	누락된 데이터를 버킷이 존재하지 않는 것으로 간주.
insert_zeros	누락된 값을 0으로 대체

// 파생 집계
{
  "derivative": {
    "buckets_path": "bytes_sum"
  }
}

// 누적 집계
{
  "cumulative_sum": {
    "buckets_path": "bytes_sum"
  }
}

// 버킷 스크립트 집계
{
  "bucket_script": {
    "buckets_path": {
      "my_var1": "bytes_sum",
      "my_var2": "total_count"
    },
    "script": "params.my_var1 / params.my_var2"
  }
}

// 버킷 셀렉터
{
  "bucket_selector": {
    "buckets_path": {
      "my_var1": "bytes_sum",
      "my_var2": "total_count"
    },
    "script": "params.my_var1 > params.my_var2"
  }
}

// 시계열 차분 집계
{
  "serial_diff": {
    "buckets_path": "bytes_sum",
    "lag": "7"
  }
}