一、了解數據及建模

北京空氣質量數據，下載地址

數據建模：

PUT air_quality {"mappings": {"doc": {"dynamic": false,"properties": {"@timestamp": {"type": "date"},"city": {"type": "keyword","ignore_above": 256},"parameter": {"type": "keyword","ignore_above": 256},"status": {"type": "keyword","ignore_above": 256},"value": {"type": "long"}}}},"settings": {"index": {"number_of_shards": "1","number_of_replicas": "0"}} }

二、導入數據

本次實驗使用Filebeat + Ingest Nodde架構。

filebeat 配置：

filebeat.prospectors: - type: logenabled: truepaths:- /home/wfs/data/*.csvexclude_lines: ["^A ","^The","^Site","^,"]output.elasticsearch:hosts: ["192.168.20.20:9200"]pipeline: "airquality"index: "air_quality"username: "elastic"password: "123456"setup.template.name: "airquality" setup.template.pattern: "airquality*" setup.template.enabled: false

數據：

# ls /home/wfs/data/ Beijing_2008_HourlyPM2.5_created20140325.csv Beijing_2014_HourlyPM25_created20150203.csv Beijing_2009_HourlyPM25_created20140709.csv Beijing_2015_HourlyPM25_created20160201.csv Beijing_2010_HourlyPM25_created20140709.csv Beijing_2016_HourlyPM25_created20170201.csv Beijing_2011_HourlyPM25_created20140709.csv Beijing_2017_HourlyPM25_created20170803.csv Beijing_2012_HourlyPM2.5_created20140325.csv Beijing_2013_HourlyPM2.5_created20140325.csv

Pipeline配置：

PUT /_ingest/pipeline/airquality {"processors": [{"grok": {"field": "message","patterns": ["%{DATA:city},%{DATA:parameter},%{DATA:date},%{NUMBER:year},%{NUMBER:month},%{NUMBER:day},%{NUMBER:hour},%{NUMBER:value},%{DATA:unit},%{DATA:duration},%{WORD:status}"]}},{"set": {"field": "_id","value": "{{city}}-{{date}}"}},{"date": {"field": "date","target_field": "@timestamp","formats": ["MM/dd/yyyy HH:mm","yyyy-MM-dd HH:mm"],"timezone": "Asia/Shanghai"}},{"remove": {"field": "message"}},{"remove": {"field": "beat"}},{"remove": {"field": "offset"}},{"remove": {"field": "source"}},{"remove": {"field": "date"}},{"convert": {"field": "year","type": "integer"}},{"convert": {"field": "month","type": "integer"}},{"convert": {"field": "day","type": "integer"}},{"convert": {"field": "hour","type": "integer"}},{"remove": {"field": "duration"}},{"remove": {"field": "unit"}},{"convert": {"field": "value","type": "integer"}}],"on_failure": [{"set": {"field": "error.message","value": "{{ _ingest.on_failure_message }}"}}] }

數據導入完成后誒下圖所示：

可以看到數據是以小時為間隔采集的，為了便于分析，可以借助python，將小時數據聚合到天的維度。

from datetime import datetime from elasticsearch import Elasticsearches = Elasticsearch(['192.168.20.20:9200'])search_query = {"query": {"range": {"value": {"gte": 1}}},"aggs": {"days": {"date_histogram": {"field": "@timestamp","interval": "day","time_zone": "+08:00"},"aggs": {"pm25": {"stats": {"field": "value"}}}}},"size": 0 } res = es.search(index='air_quality', body=search_query)index_name = 'air_quality_days' index_type = 'doc' es.indices.delete(index=index_name, ignore=[400, 404])for info in res['aggregations']['days']['buckets']:cur_date = datetime.strptime(info['key_as_string'], '%Y-%m-%dT%H:%M:%S.%f+08:00')new_doc = {"@timestamp": info['key_as_string'],'year': cur_date.year,'month': cur_date.month,'day': cur_date.day,"value_max": info['pm25']['max'],"value_avg": info['pm25']['avg'],"value_min": info['pm25']['min'],}es.index(index=index_name, doc_type=index_type, id=new_doc['@timestamp'], body=new_doc)print(new_doc)

可以看到，上邊的search_query實質上是對value大于1的按天進行分桶，并使用status返回當天PM25的系列統計值，完全等價于如下DSL：

GET air_quality/_search {"size":0,"query": {"range": {"value": {"gt": 1}}},"aggs": {"days": {"date_histogram": {"field": "@timestamp","interval": "day","time_zone": "+08:00"},"aggs": {"PM25": {"stats": {"field": "value"}}}}} }

然后對上述聚合分析循環遍歷，創建新的以天為維度的索引air_quality_days：

三、數據實戰分析

通過數據，我們首先可以從整體上看下十年來空氣質量是否有好轉：
1.空氣質量分析 – 每年藍天占比餅圖：

這里使用了腳本動態生成到了rate_level字段，

在ManageMent–>Index Patterns中配置：

腳本內容：

double val=doc['value_max'].value; String rtn=""; if(val<50){ rtn="1-Good" }else if(val<100){ rtn="2-Moderate" }else if(val<150){ rtn="3-Unhealthy for Sensitive Groups" }else if(val<200){ rtn="4-Unhealthy" }else if(val<300){ rtn="5-Very Unhealthy" }else{ rtn="6-Hazardous" } return rtn;

2.空氣質量分析 – AQI質量隨時間分布占比：

配置比較簡單：Options中設置：Chart Type：Bar Stacked：Percent

3.空氣質量分析 – 每年藍天占比（VB）：
以value_max的值等于100為判斷依據，小于100定為Good，Panel Options 中將書劍間隔Interval設置為1y即可

4.空氣質量分析 – 每月藍天占比（VB）：
同理，將時間間隔修改為1M：

然后在具體到某一時間段內，看一下空氣質量是否有改善。比如通過數據對比2016年冬季較2015年同一時刻的空氣質量情況。
1.空氣質量分析 – 2016 vs 2015 冬季霧霾天數占比（TL）

.es(index=air_quality_days,q='value_max:>150',offset=-1y).divide(.es(index=air_quality_days,offset=-1y)).multiply(100).label(2015).lines(fill=1,width=1),.es(index=air_quality_days,q='value_max:>150').divide(.es(index=air_quality_days)).multiply(100).label(2016).lines(fill=3,width=1).title("2016 vs 2015霧霾天數占比")

2.空氣質量分析 – 2016 VS 2015 PM25最大值比較 （VB）

將 Offset series time設置為1y即可得到2015年的數據，通過Fill（0-1）和Line Width控制線的樣式和透明度

3.空氣質量分析 – 2016、2015年冬季霧霾天數（VB Metric）

4.空氣質量分析 – 2016、2015年冬季每天空氣質量情況

最后附上兩個dashboard：

總結

以上是生活随笔為你收集整理的ELK实践（三）北京历年空气质量数据分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。