在之前的教程 “Elasticsearch：pipeline aggregation 介绍” 中，我介绍了一种 pipeline aggregation。里面有介绍一种叫做 derivative aggregation 的。先前的 derivative 可视化将存储桶与相邻存储桶进行了比较，这是日常有用的观察。 但是这些存储桶仅将一天与第二个存储桶进行比较，理想情况下，我们可能希望比较一周中某一天的存储桶，例如比较一个星期一与下一个星期一。 为此，我们将需要使用 Serial Diff 管道聚合。在实际的应用中，比如我们统计每周一和上一个周一的车流量比较；今年一月份和去年一月份的财报比较。
Serial Diff 是一种在不同的 lag 或周期从其自身中减去时间序列中的值的技术。 例如，数据点 f(x）= f(xt) - f(xt-n)，其中 n 是使用的周期。
周期1等于没有时间归一化的导数：它只是从一个点到另一个点的变化。 单周期对于消除恒定的线性趋势很有用。
在今天的练习中，我们来展示如何使用 Serial Diff 来对我们的周期性的数据进行统计。
准备数据
首先，我们在 Kibana 中打入如下的命令：
POST sales/_bulk
{ "index" : { "_id" : "1" } }
{ "date" : "2020-09-01", "value": 1 }
{ "index" : { "_id" : "2" } }
{ "date" : "2020-09-02", "value": 2 }
{ "index" : { "_id" : "3" } }
{ "date" : "2020-09-03", "value": 3 }
{ "index" : { "_id" : "4" } }
{ "date" : "2020-09-04", "value": 4 }
{ "index" : { "_id" : "5" } }
{ "date" : "2020-09-05", "value": 5 }
{ "index" : { "_id" : "6" } }
{ "date" : "2020-09-06", "value": 6 }
{ "index" : { "_id" : "7" } }
{ "date" : "2020-09-07", "value": 7 }
{ "index" : { "_id" : "8" } }
{ "date" : "2020-09-08", "value": 8 }
{ "index" : { "_id" : "9" } }
{ "date" : "2020-09-09", "value": 9 }
{ "index" : { "_id" : "10" } }
{ "date" : "2020-09-10", "value": 10 }
{ "index" : { "_id" : "11" } }
{ "date" : "2020-09-11", "value": 11 }
{ "index" : { "_id" : "12" } }
{ "date" : "2020-09-12", "value": 12 }
{ "index" : { "_id" : "13" } }
{ "date" : "2020-09-13", "value": 13 }
{ "index" : { "_id" : "14" } }
{ "date" : "2020-09-14", "value": 14 }
{ "index" : { "_id" : "15" } }
{ "date" : "2020-09-15", "value": 15 }
{ "index" : { "_id" : "16" } }
{ "date" : "2020-09-16", "value": 16 }
{ "index" : { "_id" : "17" } }
{ "date" : "2020-09-17", "value": 17 }
{ "index" : { "_id" : "18" } }
{ "date" : "2020-09-18", "value": 18 }
{ "index" : { "_id" : "19" } }
{ "date" : "2020-09-19", "value": 19 }
{ "index" : { "_id" : "20" } }
{ "date" : "2020-09-20", "value": 20 }
{ "index" : { "_id" : "21" } }
{ "date" : "2020-09-21", "value": 21 }
{ "index" : { "_id" : "22" } }
{ "date" : "2020-09-22", "value": 22 }
{ "index" : { "_id" : "23" } }
{ "date" : "2020-09-23", "value": 23 }
{ "index" : { "_id" : "24" } }
{ "date" : "2020-09-24", "value": 24 }
{ "index" : { "_id" : "25" } }
{ "date" : "2020-09-25", "value": 25 }
{ "index" : { "_id" : "26" } }
{ "date" : "2020-09-26", "value": 26 }
{ "index" : { "_id" : "27" } }
{ "date" : "2020-09-27", "value": 27 }
{ "index" : { "_id" : "28" } }
{ "date" : "2020-09-28", "value": 28 }
{ "index" : { "_id" : "29" } }
{ "date" : "2020-09-29", "value": 29 }
{ "index" : { "_id" : "30" } }
{ "date" : "2020-09-30", "value": 30 }上面的 _bulk 指令可以帮我们一次性地把数据导入到 Elasticsearch 中。索引 sales 是一个假想的销售记录。在一个月刚开始的第一天，销售额是1，到月底是30。
运用 Serial Diff 来对数据进行统计
我们在 Kibana 中，通过 Stack Management => Index Patterns 来对我们的 sales 索引创建一个叫做 sales* 的 index pattern。接着我们使用 Visualize 来对我们的数据可视化：
我们首先设定 date histogram bucket，因为这是一个时序的数据：
接着配置上面的 Metrics 部分：
在这里，我们选择 Serial Diff aggregation，同时我们在 Custome metric 部分，选择 Sum，并针对 value 这个字段。我们点击 Advanced，并使用 JSON 格式：
{"lag":7}来对数据的周期进行设定。这样的配置显示的是周期是一个星期。由于我们的数据是线性的，所以每隔一个星期的差值就是7，而且差值都是一样的。如果我们设定：
{"lag":4}
我们可以从上面看出现在的差值变为4了。当然，如果我们设置 lag 为1，那么就是之前的 derivative aggregation了。
假如大家想知道这个请求是什么样子的，你可以通过如下的方式来获得：
在上面罗列了整个数据的请求格式。我们可以点击右边的拷贝按钮，把请求的内容拷贝下来，并粘贴到 Dev Tools 中。我们添加 GET sales*/_search：
GET sales/_search
{"aggs": {"2": {"date_histogram": {"field": "date","calendar_interval": "1d","time_zone": "Asia/Shanghai"},"aggs": {"1": {"serial_diff": {"buckets_path": "1-metric","lag": 4}},"1-metric": {"sum": {"field": "value"}}}}},"size": 0,"stored_fields": ["*"],"script_fields": {},"docvalue_fields": [{"field": "date","format": "date_time"}],"_source": {"excludes": []},"query": {"bool": {"must": [],"filter": [{"match_all": {}},{"range": {"date": {"gte": "2020-09-03T03:45:38.861Z","lte": "2020-10-13T03:45:38.861Z","format": "strict_date_optional_time"}}}],"should": [],"must_not": []}}
}上面将显示完整的请求命令。