Aggregating data for faster performance

By default, datafeeds fetch data from Elasticsearch using search and scroll requests. It can be significantly more efficient, however, to aggregate data in Elasticsearch and to configure your anomaly detection jobs to analyze aggregated data.

One of the benefits of aggregating data this way is that Elasticsearch automatically distributes these calculations across your cluster. You can then feed this aggregated data into the machine learning features instead of raw results, which reduces the volume of data that must be considered while detecting anomalies.

There are some limitations to using aggregations in datafeeds.

Your aggregation must include a date_histogram aggregation or a top level composite aggregation, which in turn must contain a max aggregation on the time field. This requirement ensures that the aggregated data is a time series and the timestamp of each bucket is the time of the last record in the bucket.

You must consider the interval of the date_histogram or composite aggregation carefully. The bucket span of your anomaly detection job must be divisible by the value of the calendar_interval or fixed_interval in your aggregation (with no remainder). If you specify a frequency for your datafeed, it must also be divisible by this interval. Anomaly detection jobs cannot use date_histogram or composite aggregations with an interval measured in months because the length of the month is not fixed; they can use weeks or smaller units.

If your datafeed uses aggregations with nested terms aggs and model plot is not enabled for the anomaly detection job, neither the Single Metric Viewer nor the Anomaly Explorer can plot and display an anomaly chart for the job. In these cases, the charts are not visible and an explanatory message is shown.

Your datafeed can contain multiple aggregations, but only the ones with names that match values in the job configuration are fed to the job.

When you create or update an anomaly detection job, you can include the names of aggregations, for example:

PUT _ml/anomaly_detectors/farequote
{
  "analysis_config": {
    "bucket_span": "60m",
    "detectors": [{
      "function": "mean",
      "field_name": "responsetime",  
      "by_field_name": "airline"  
    }],
    "summary_count_field_name": "doc_count"
  },
  "data_description": {
    "time_field":"time"  
  }
}

The aggregations are defined in the datafeed as follows:

PUT _ml/datafeeds/datafeed-farequote
{
  "job_id":"farequote",
  "indices": ["farequote"],
  "aggregations": {
    "buckets": {
      "date_histogram": {
        "field": "time",
        "fixed_interval": "360s",
        "time_zone": "UTC"
      },
      "aggregations": {
        "time": {  
          "max": {"field": "time"}
        },
        "airline": {  
          "terms": {
            "field": "airline",
            "size": 100
          },
          "aggregations": {
            "responsetime": {  
              "avg": {
                "field": "responsetime"
              }
            }
          }
        }
      }
    }
  }
}

For composite aggregation support, there must be exactly one date_histogram value source. That value source must not be sorted in descending order. Additional composite aggregation value sources are allowed, such as terms.

Here is an example that uses a composite aggregation instead of a date_histogram.

Assuming the same job configuration as above.

PUT _ml/anomaly_detectors/farequote-composite
{
  "analysis_config": {
    "bucket_span": "60m",
    "detectors": [{
      "function": "mean",
      "field_name": "responsetime",
      "by_field_name": "airline"
    }],
    "summary_count_field_name": "doc_count"
  },
  "data_description": {
    "time_field":"time"
  }
}

This is an example of a datafeed that uses a composite aggregation to bucket the metrics based on time and terms:

PUT _ml/datafeeds/datafeed-farequote-composite
{
  "job_id": "farequote-composite",
  "indices": [
    "farequote"
  ],
  "aggregations": {
    "buckets": {
      "composite": {
        "size": 1000,  
        "sources": [
          {
            "time_bucket": {  
              "date_histogram": {
                "field": "time",
                "fixed_interval": "360s",
                "time_zone": "UTC"
              }
            }
          },
          {
            "airline": {  
              "terms": {
                "field": "airline"
              }
            }
          }
        ]
      },
      "aggregations": {
        "time": {  
          "max": {
            "field": "time"
          }
        },
        "responsetime": { 
          "avg": {
            "field": "responsetime"
          }
        }
      }
    }
  }
}

Datafeeds support complex nested aggregations. This example uses the derivative pipeline aggregation to find the first order derivative of the counter system.network.out.bytes for each value of the field beat.name.

"aggregations": {
  "beat.name": {
    "terms": {
      "field": "beat.name"
    },
    "aggregations": {
      "buckets": {
        "date_histogram": {
          "field": "@timestamp",
          "fixed_interval": "5m"
        },
        "aggregations": {
          "@timestamp": {
            "max": {
              "field": "@timestamp"
            }
          },
          "bytes_out_average": {
            "avg": {
              "field": "system.network.out.bytes"
            }
          },
          "bytes_out_derivative": {
            "derivative": {
              "buckets_path": "bytes_out_average"
            }
          }
        }
      }
    }
  }
}

Datafeeds not only supports multi-bucket aggregations, but also single bucket aggregations. The following shows two filter aggregations, each gathering the number of unique entries for the error field.

{
  "job_id":"servers-unique-errors",
  "indices": ["logs-*"],
  "aggregations": {
    "buckets": {
      "date_histogram": {
        "field": "time",
        "interval": "360s",
        "time_zone": "UTC"
      },
      "aggregations": {
        "time": {
          "max": {"field": "time"}
        }
        "server1": {
          "filter": {"term": {"source": "server-name-1"}},
          "aggregations": {
            "server1_error_count": {
              "value_count": {
                "field": "error"
              }
            }
          }
        },
        "server2": {
          "filter": {"term": {"source": "server-name-2"}},
          "aggregations": {
            "server2_error_count": {
              "value_count": {
                "field": "error"
              }
            }
          }
        }
      }
    }
  }
}

When you define an aggregation in a datafeed, it must have one of the following forms:

When using a date_histogram aggregation to bucket by time:

"aggregations": {
  ["bucketing_aggregation": {
    "bucket_agg": {
      ...
    },
    "aggregations": {]
      "data_histogram_aggregation": {
        "date_histogram": {
          "field": "time",
        },
        "aggregations": {
          "timestamp": {
            "max": {
              "field": "time"
            }
          },
          [,"<first_term>": {
            "terms":{...
            }
            [,"aggregations" : {
              [<sub_aggregation>]+
            } ]
          }]
        }
      }
    }
  }
}

When using a composite aggregation:

"aggregations": {
  "composite_agg": {
    "sources": [
      {
        "date_histogram_agg": {
          "field": "time",
          ...settings...
        }
      },
      ...other valid sources...
      ],
      ...composite agg settings...,
      "aggregations": {
        "timestamp": {
            "max": {
              "field": "time"
            }
          },
          ...other aggregations...
          [
            [,"aggregations" : {
              [<sub_aggregation>]+
            } ]
          }]
      }
   }
}

The top level aggregation must be exclusively one of the following:

There must be exactly one date_histogram, composite aggregation. For more information, see Date histogram aggregation and Composite aggregation.

Each histogram or composite bucket has a key, which is the bucket start time. This key cannot be used for aggregations in datafeeds, however, because they need to know the time of the latest record within a bucket. Otherwise, when you restart a datafeed, it continues from the start time of the histogram or composite bucket and possibly fetches the same data twice. The max aggregation for the time field is therefore necessary to provide the time of the latest record within a bucket.

You can optionally specify a terms aggregation, which creates buckets for different values of a field.

You can change this behavior by setting the size parameter. To determine the cardinality of your data, you can run searches such as:

GET .../_search
{
  "aggs": {
    "service_cardinality": {
      "cardinality": {
        "field": "service"
      }
    }
  }
}

By default, Elasticsearch limits the maximum number of terms returned to 10000. For high cardinality fields, the query might not run. It might return errors related to circuit breaking exceptions that indicate that the data is too large. In such cases, use composite aggregations in your datafeed. For more information, see Terms aggregation.

You can also optionally specify multiple sub-aggregations. The sub-aggregations are aggregated for the buckets that were created by their parent aggregation. For more information, see Aggregations.