Elasticsearchにおける集計処理の実装パターン

Elasticsearchでは、text型フィールドに対してデフォルトで集計やソートが禁止されています。fielddata=trueを設定することで有効化できます。


PUT access_logs/_mapping
{
  "properties": {
    "publisher_org": {
      "type": "text",
      "fielddata": true
    }
  }
}

1. 基本的なバケット集計

単純なグループ化によるドキュメント数の集計です。

クエリ例


GET access_logs/_search
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "terms": {
            "site_code": ["site_001"]
          }
        },
        {
          "range": {
            "publish_date": {
              "gte": "2023-01-01",
              "lte": "2023-12-31"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "category_groups": {
      "terms": {
        "field": "category_id",
        "size": 100,
        "order": [
          {"_count": "asc"},
          {"_key": "asc"}
        ]
      }
    }
  }
}

Java実装例


SearchSourceBuilder queryBuilder = new SearchSourceBuilder();
queryBuilder.size(0);

BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must(QueryBuilders.termsQuery("site_code", siteCodes));
boolQuery.must(QueryBuilders.rangeQuery("publish_date")
    .from(startDate).to(endDate));

TermsAggregationBuilder categoryAgg = AggregationBuilders
    .terms("category_groups")
    .field("category_id")
    .size(100)
    .order(BucketOrder.count(true));

queryBuilder.query(boolQuery).aggregation(categoryAgg);

SearchRequest request = new SearchRequest("access_logs").source(queryBuilder);
SearchResponse response = client.search(request, RequestOptions.DEFAULT);

if (response.status() == RestStatus.OK) {
    ParsedStringTerms terms = response.getAggregations().get("category_groups");
    for (Terms.Bucket bucket : terms.getBuckets()) {
        CategoryStats stats = new CategoryStats();
        stats.setId(bucket.getKeyAsString());
        stats.setCount(bucket.getDocCount());
        resultList.add(stats);
    }
}

2. サブ集計を伴うバケット集計

グループ化した上で、各バケットに対してさらに集計を実行します。ここではユニークIP数の算出を例示します。

クエリ例


GET access_logs/_search
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "terms": {
            "site_code": ["site_001"]
          }
        },
        {
          "range": {
            "access_timestamp": {
              "gte": "2023-01-01",
              "lte": "2023-12-31"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "category_groups": {
      "terms": {
        "field": "category_id",
        "size": 100,
        "order": [
          {"_count": "asc"},
          {"_key": "asc"}
        ]
      },
      "aggs": {
        "unique_visitors": {
          "cardinality": {
            "field": "client_ip"
          }
        }
      }
    }
  }
}

Java実装例


SearchSourceBuilder queryBuilder = new SearchSourceBuilder();
queryBuilder.size(0);

BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must(QueryBuilders.termsQuery("site_code", siteCodes));
boolQuery.must(QueryBuilders.rangeQuery("access_timestamp")
    .from(startDate).to(endDate));

TermsAggregationBuilder categoryAgg = AggregationBuilders
    .terms("category_groups")
    .field("category_id")
    .size(100)
    .order(BucketOrder.count(true))
    .subAggregation(
        AggregationBuilders.cardinality("unique_visitors")
            .field("client_ip")
    );

queryBuilder.query(boolQuery).aggregation(categoryAgg);

SearchRequest request = new SearchRequest("access_logs").source(queryBuilder);
SearchResponse response = client.search(request, RequestOptions.DEFAULT);

if (response.status() == RestStatus.OK) {
    ParsedStringTerms terms = response.getAggregations().get("category_groups");
    for (Terms.Bucket bucket : terms.getBuckets()) {
        ParsedCardinality uniqueVisitors = bucket.getAggregations().get("unique_visitors");
        
        AccessStats stats = new AccessStats();
        stats.setCategoryId(bucket.getKeyAsString());
        stats.setPageViews(bucket.getDocCount());
        stats.setUniqueVisitors(uniqueVisitors.getValue());
        resultList.add(stats);
    }
}

3. 多階層バケット集計

複数のフィールドで階層的にグループ化を行います。時間軸とクライアントIPでの集計例です。

クエリ例


GET access_logs/_search
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "term": {
            "site_code": "site_001"
          }
        },
        {
          "range": {
            "access_timestamp": {
              "gte": "2023-01-01",
              "lte": "2023-12-31"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "time_groups": {
      "terms": {
        "field": "year_month",
        "size": 10000,
        "order": {"_key": "asc"}
      },
      "aggs": {
        "ip_groups": {
          "terms": {
            "field": "client_ip",
            "size": 10000,
            "order": [
              {"_count": "desc"},
              {"_key": "asc"}
            ]
          },
          "aggs": {
            "unique_visitors": {
              "cardinality": {
                "field": "client_ip"
              }
            }
          }
        }
      }
    }
  }
}

Java実装例


SearchSourceBuilder queryBuilder = new SearchSourceBuilder();
queryBuilder.size(0);

BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must(QueryBuilders.termQuery("site_code", siteCode));
boolQuery.must(QueryBuilders.rangeQuery("access_timestamp")
    .from(startDate).to(endDate));

String timeField = resolveTimeField(dateType); // year_month, year_month_day, etc.

TermsAggregationBuilder ipAgg = AggregationBuilders
    .terms("ip_groups")
    .field("client_ip")
    .size(10000)
    .order(BucketOrder.count(false))
    .subAggregation(
        AggregationBuilders.cardinality("unique_visitors")
            .field("client_ip")
    );

TermsAggregationBuilder timeAgg = AggregationBuilders
    .terms("time_groups")
    .field(timeField)
    .size(10000)
    .order(BucketOrder.key(true))
    .subAggregation(ipAgg);

queryBuilder.query(boolQuery).aggregation(timeAgg);

SearchRequest request = new SearchRequest("access_logs").source(queryBuilder);
SearchResponse response = client.search(request, RequestOptions.DEFAULT);

if (response.status() == RestStatus.OK) {
    ParsedStringTerms timeTerms = response.getAggregations().get("time_groups");
    for (Terms.Bucket timeBucket : timeTerms.getBuckets()) {
        String timeKey = timeBucket.getKeyAsString();
        ParsedStringTerms ipTerms = timeBucket.getAggregations().get("ip_groups");
        
        TimeAccessStats timeStats = new TimeAccessStats();
        timeStats.setTimePeriod(formatTimeKey(timeKey, dateType));
        timeStats.setTotalAccess(timeBucket.getDocCount());
        
        for (Terms.Bucket ipBucket : ipTerms.getBuckets()) {
            ParsedCardinality uniqueVisitors = ipBucket.getAggregations().get("unique_visitors");
            
            IpAccessStats ipStats = new IpAccessStats();
            ipStats.setIpAddress(ipBucket.getKeyAsString());
            ipStats.setAccessCount(ipBucket.getDocCount());
            ipStats.setUniqueVisitors(uniqueVisitors.getValue());
            timeStats.addIpDetail(ipStats);
        }
        resultList.add(timeStats);
    }
}

4. 日時ヒストグラム集計

日付フィールドに対して、指定した間隔でヒストグラムを生成します。

間隔別設定例

日単位:


"aggs": {
  "daily_stats": {
    "date_histogram": {
      "field": "access_timestamp",
      "calendar_interval": "day",
      "format": "yyyy-MM-dd",
      "min_doc_count": 0
    }
  }
}

月単位:


"aggs": {
  "monthly_stats": {
    "date_histogram": {
      "field": "access_timestamp",
      "calendar_interval": "month",
      "format": "yyyy-MM",
      "min_doc_count": 0
    }
  }
}

年単位:


"aggs": {
  "yearly_stats": {
    "date_histogram": {
      "field": "access_timestamp",
      "calendar_interval": "year",
      "format": "yyyy",
      "min_doc_count": 0
    }
  }
}

Java実装例


DateHistogramInterval interval;
String format;
String start;
String end;

switch (dateType) {
    case 1: // 日別
        interval = DateHistogramInterval.DAY;
        format = "yyyy-MM-dd";
        start = "2023-10-01";
        end = "2023-10-31";
        break;
    case 2: // 月別
        interval = DateHistogramInterval.MONTH;
        format = "yyyy-MM";
        start = "2022-09";
        end = "2023-10";
        break;
    default: // 年別
        interval = DateHistogramInterval.YEAR;
        format = "yyyy";
        start = "2022";
        end = "2023";
}

DateHistogramAggregationBuilder dateAgg = AggregationBuilders
    .dateHistogram("date_stats")
    .field("access_timestamp")
    .calendarInterval(interval)
    .format(format)
    .minDocCount(0)
    .extendedBounds(new LongBounds(start, end));

5. 統計指標の取得

stats集計を使用すると、フィールドの最小値、最大値、平均値、合計、件数を一括で取得できます。

6. 合計値の算出

sum集計を使用して、数値フィールドの合計値を算出します。

タグ: Elasticsearch Aggregation Java API date_histogram terms aggregation

7月13日 22:46 投稿