es數據二次開發統計展示

案例1

      在es查詢中按照多列分組的時候 分組列的count值會越來越少 es默認隱藏了沒有被分組匹配到的記錄數  需要在查詢的時候開啟

 

 2.開啟顯示沒有被分組成功的記錄

    分組成功的記錄加上分組missing的記錄數就等於總的記錄數  26932+2666=29598

 3.當實際的總數和es分組統計的條數對不上的時候 需要考慮是不是分組列的值有可能被丟失了 這個時候可以開啟顯示丟失

4.查看es的原始日志內容確實有10001條記錄不存在CHANNEL字段

 

實例統計

#!/usr/bin/env python
# -*- coding: utf-8 -*-


from elasticsearch6 import Elasticsearch
import datetime
import time
import re

es = Elasticsearch("http://10.000.142.88:9200")

#每小時定時執行統計前一個小時的數據
def formartTime(startTime):
    try:
        startTime = datetime.datetime.strptime(startTime, '%Y-%m-%dT%H:%M:%S.%f') 
    except Exception as e:
        startTime = datetime.datetime.strptime(startTime, '%Y-%m-%d %H:%M:%S') 
    startTime = startTime.strftime('%Y-%m-%d %H:%M:%S.%f')[:-13]
    return startTime+":00:00"

def strtime_to_datetime(timestr):
    """將字符串格式的時間 (含毫秒) 轉為 datetime 格式
    :param timestr: {str}'2016-02-25 20:21:04.242'
    :return: {datetime}2016-02-25 20:21:04.242000
    """
    local_datetime = datetime.datetime.strptime(timestr, "%Y-%m-%d %H:%M:%S.%f")
    return local_datetime

def datetime_to_timestamp(datetime_obj):
    """將本地(local) datetime 格式的時間 (含毫秒) 轉為毫秒時間戳
    :param datetime_obj: {datetime}2016-02-25 20:21:04.242000
    :return: 13 位的毫秒時間戳  1456402864242
    """
    local_timestamp = int(time.mktime(datetime_obj.timetuple()) * 1000.0 + datetime_obj.microsecond / 1000.0)
    return local_timestamp

def strtime_to_timestamp(local_timestr):
    """將本地時間 (字符串格式，含毫秒) 轉為 13 位整數的毫秒時間戳
    :param local_timestr: {str}'2016-02-25 20:21:04.242'
    :return: 1456402864242
    """
    local_datetime = strtime_to_datetime(local_timestr)
    timestamp = datetime_to_timestamp(local_datetime)
    return timestamp


today=datetime.date.today()
tnow=datetime.datetime.now()

startTime=(datetime.datetime.now()+datetime.timedelta(hours=-3)).replace(minute=0,second=0).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
endTime=(datetime.datetime.now()+datetime.timedelta(hours=-3)).replace(minute=59,second=59).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]


stime=str(strtime_to_timestamp(startTime))[:-3]+"000"
etime=str(strtime_to_timestamp(endTime))[:-3]+"999"


def getindex():
  if tnow.hour>2:
    indexname=today.strftime("%Y-%m-%d")
  else:
    indexname=(tnow+datetime.timedelta(days=-1)).strftime("%Y-%m-%d")
  return indexname

indexname="sage-send-"+str(today)
findexname="as*"+getindex()


body={"aggs":{"2":{"terms":{"field":"APP_ID","size":5000,"order":{"_count":"desc"}},"aggs":{"3":{"terms":{"field":"CHANNEL","size":5000,"order":{"_count":"desc"}},"aggs":{"4":{"terms":{"field":"CHANNEL_ID","size":5000,"order":{"_count":"desc"}},"aggs":{"5":{"terms":{"field":"SWJG_DM","size":5000,"order":{"_count":"desc"}},"aggs":{"6":{"terms":{"field":"MESSAGE_TYPE","size":5000,"order":{"_count":"desc"}},"aggs":{"7":{"date_histogram":{"field":"mydate","interval":"1h","time_zone":"Asia/Shanghai","min_doc_count":1}}}}}}}}}}}}},"size":0,"_source":{"excludes":[]},"stored_fields":["*"],"script_fields":{},"docvalue_fields":[{"field":"@timestamp","format":"date_time"},{"field":"mydate","format":"date_time"}],"query":{"bool":{"must":[{"match_phrase":{"metricsName":{"query":"消息發送量統計"}}},{"match_all":{}},{"range":{"mydate":{"gte":stime,"lte":etime,"format":"epoch_millis"}}}],"filter":[],"should":[],"must_not":[]}},"timeout":"30000ms"}

if es.indices.exists(index=findexname):
    res = es.search(body=body,index=findexname)
    outlist=[]
    dnow=datetime.datetime.now().strftime('%Y-%m-%d %H')+":00:00"
    for i2 in res["aggregations"]["2"]["buckets"]:
        for i3 in i2["3"]["buckets"]:
          for i4 in i3["4"]["buckets"]:
            for i5 in i4["5"]["buckets"]:
              for i6 in i5["6"]["buckets"]:
                for i7 in i6["7"]["buckets"]:
                  timestr = i7["key_as_string"][:-6]
                  newtime = formartTime(timestr)
                  outlist.append({"appId":i2["key"],"count":i7["doc_count"],"channel":i3["key"],"channelId":i4["key"],"swjgDm":i5["key"],"messageType":i6["key"],"creatTime":newtime,"statisticalTime":dnow})

    if es.indices.exists(index=indexname):
        pass
    else:
        es.indices.create(index=indexname)

    for data in outlist:
        res = es.index(index=indexname, doc_type="doc", body=data)

 

discover面板

 

如何查看指定索引名稱的創建時間

     1.命令行查詢

      curl -XGET http://192.168.80.10:9200/zhouls/_settings?pretty 

      {
        "zhouls" : {
             "settings" : {
                 "index" : {
                         "creation_date" : "1488203759467",  //表示索引的創建時間
                         "uuid" : "Sppm-db_Qm-OHptOC7vznw",
                         "number_of_replicas" : "1",
                         "number_of_shards" : "5",
                         "version" : {
                              "created" : "2040399"
          }
    }

2.通過kibana查看索引的創建時間