Python Elasticsearch API操作ES集群
【摘要】 环境Centos 7.4Python 2.7Pip 2.7 MySQL-python 1.2.5 Elasticsearc 6.3.1Elasitcsearch6.3.2知识点调用Python Elasticsearh APIPython Mysqldb使用DSL查询与聚合Python 列表操作代码#!/usr/bin/env python# -*- coding: utf-8 -*-#mi...
环境
Centos 7.4
Python 2.7
Pip 2.7 MySQL-python 1.2.5 Elasticsearc 6.3.1
Elasitcsearch6.3.2
知识点
调用Python Elasticsearh API
Python Mysqldb使用
DSL查询与聚合
Python 列表操作
代码
#!/usr/bin/env python# -*- coding: utf-8 -*-#minyt 2018.9.1#获取24小时内出现的模块次数# 该程序通过elasticsearch python client 获取相关精简数据,可以计算请求数、超时数、错误数、正确率、错误率等等import MySQLdbfrom elasticsearch import Elasticsearchfrom elasticsearch import helpers#定义elasticsearch集群索引名index_name = "logstash-nginxlog-*"#实例化Elasticsearch类,并设置超时间为180秒,默认是10秒的,如果数据量很大,时间设置更长一些es = Elasticsearch(['elasticsearch01','elasticsearch02','elasticsearch03'],timeout=180)#DSL(领域特定语言)查询语法,查询top50 sname的排列次数data_sname = { "aggs": { "2": { "terms": { "field": "apistatus.sname.keyword", "size": 100, "order": { "_count": "desc"
}
}
}
}, "size": 0, "_source": { "excludes": []
}, "stored_fields": [ "*"
], "script_fields": {}, "docvalue_fields": [ "@timestamp"
], "query": { "bool": { "must": [
{ "match_all": {}
},
{ "range": { "@timestamp": { "gte" : "now-24h/h", "lt" : "now/h"
}
}
}
], "filter": [], "should": [], "must_not": []
}
}
}#按照DSL(特定领域语言)语法查询获取数据def get_original_data():
try: #根据上面条件搜索数据
res = es.search(
index=index_name,
size=0,
body=data_sname
) return res except: print "get original data failure"#初始化数据库def init_mysql():
# 打开数据库连接
db = MySQLdb.connect("localhost", "myuser", "mypassword", "mydb", charset='utf8' ) # 使用cursor()方法获取操作游标
cursor = db.cursor() # SQL 更新语句
sql = "update appname set count=0"
try: # 执行SQL语句
cursor.execute(sql) # 提交到数据库执行
db.commit() except: # 发生错误时回滚
db.rollback() # 关闭数据库连接
db.close()def updata_mysql(sname_count,sname_list):
# 打开数据库连接
db = MySQLdb.connect("localhost", "myuser", "mypassword", "mydb", charset='utf8' ) # 使用cursor()方法获取操作游标
cursor = db.cursor() # SQL 更新语句
sql = "update appname set count=%d where sname = '%s'" % (sname_count,sname_list) try: # 执行SQL语句
cursor.execute(sql) # 提交到数据库执行
db.commit() except: # 发生错误时回滚
db.rollback() # 关闭数据库连接
db.close()#根据Index数据结构通过Elasticsearch Python Client上传数据到新的Indexdef import_process_data():
try: #列表形式显示结果
res = get_original_data() #print res
res_list = res.get('aggregations').get('2').get('buckets') #print res_list
#初始化数据库
init_mysql() #获取24小时内出现的SNAME
for value in res_list:
sname_list = value.get('key')
sname_count = value.get('doc_count') print sname_list,sname_count #更新sname_status值
updata_mysql(sname_count,sname_list) except Exception, e: print repr(e)if __name__ == "__main__":
import_process_data()
总结
关键是DSL语法的编写涉及查询与聚合可以通过kibana的visualize或者devtool先测试出正确语法,然后结合python对列表、字典、除法、字符串等操作即可。下面汇总下各个算法:
总请求
http_host.keyword: api.mydomain.com超长请求
http_host.keyword: api.mydomain.com AND request_time: [1 TO 600] NOT apistatus.status.keyword:*错误错误请求
apistatus.status.keyword:*错误 AND (http_host.keyword: api.mydomain.com OR http_host.keyword: api.yourdomain.com )请求健康度
域名与request_time聚合,域名请求时间小于3秒的次数除以总请求次数对应各个域名健康度请求正确率
域名与http状态码聚合,域名http状态码为200的次数除以域名总请求数对应各个域名的请求正确率
本文转自minminmsn博客51CTO博客,如需转载,请自行联系原作者。
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)