容器化部署elasticsearch教程+python操作es数据库示例

1. 拉取镜像dockerpull elasticsearch:7.17.12. 创建配置文件mkdir-p/home/elasticsearchsudochmod777/home/elasticsearchmkdir-p/home/elasticsearch/configmkdir-p/home/elasticsearch/homemkdir-p/home/elasticsearch/logsmkdir-p/home/elasticsearch/datamkdir-p/home/elasticsearch/pluginsechohttp.host: 0.0.0.0/home/elasticsearch/config/elasticsearch.ymlchown-R1000:1000 /home/elasticsearch/logs /home/elasticsearch/data3. 创建容器dockerrun-itd-p32140:9200-p32139:9300\--restartalways\--privilegedtrue\--nameelasticsearch-filebeat\-ediscovery.typesingle-node-eES_JAVA_OPTS-Xms16g -Xmx16g\-ediscovery.typesingle-node\-eELASTIC_PASSWORDpaas123\-expack.security.enabledtrue\-v/home/elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml\-v/home/elasticsearch/logs:/usr/share/elasticsearch/logs\-v/home/elasticsearch/data:/usr/share/elasticsearch/data\-v/home/elasticsearch/home:/usr/share/elasticsearch/home\-v/home/elasticsearch/plugins:/usr/share/elasticsearch/plugins\elasticsearch:7.17.14. 验证服务[rootlocalhost elasticsearch]# curl -k http://localhost:32140/_cat/health?v --user elastic:paas123epoch timestamp cluster status node.total node.data shards pri relo init unassign pending_tasks max_task_wait_time active_shards_percent177632939708:49:57 elasticsearch yellow11110010-50.0%[rootlocalhost elasticsearch]#5. web api接口示例ES教程# 查询数据http://10.132.1.126:32140/{index_name}/_search?from0size100timeout1s# 通过用例名称查询数据http://10.132.1.126:32140/{index_name}/_search?qcase_name:用户查看云主机权限策略资源粒度下的告警日志生成from0size10timeout1s6. python 操作ES示例脚本# -*- coding: utf-8 -*-# Time : 2026/4/15 15:11# Software: PyCharm# Desc :importhashlibimportwarningsfromelasticsearchimportElasticsearch# Elasticsearch7.17.1服务版本与python Elasticsearch7.17.5库版本不要差异太大不同版本有差异# 精准匹配这个废弃警告忽略它warnings.filterwarnings(ignore,categoryDeprecationWarning,messagerThe body parameter is deprecated for the search API and will be removed in a future version.*)# ES 相关配置ES_HOST10.132.1.126ES_PORT32140ES_USERelasticES_PWDpaas123# 定义 Mappingindex_nametest_api_mapsmapping_body{mappings:{# 【新增】全局模板将所有字符串字段强制映射为 keyword防止空数组被识别为 textdynamic_templates:[{strings_as_keyword:{match_mapping_type:string,mapping:{type:keyword,ignore_above:256}}}],properties:{# --- 基础信息 ---case_path:{type:keyword,# 用例路径用于精确查找和去重ignore_above:256# 超过256字符不索引防止超长报错},case_name:{type:keyword,# 用例名称ignore_above:256},# --- 核心嵌套结构 ---api_tree:{type:nested,# 【关键】必须是 nested因为它是对象数组properties:{# 1. 前置明细 (动态对象)prefix_step:{type:nested,dynamic:True,# 允许动态添加步骤名如 创建告警策略properties:{step_name:{type:keyword},# 存储步骤名details:{# 存储接口数组type:nested,properties:{method:{type:keyword},url:{type:keyword},status:{type:keyword}}}}},# 2. 步骤明细 (动态对象)case_step:{type:nested,dynamic:True,# 允许动态添加步骤名如 创建告警策略properties:{step_name:{type:keyword},# 存储步骤名details:{# 存储接口数组type:nested,properties:{method:{type:keyword},url:{type:keyword},status:{type:keyword}}}}},# 3. 接口汇总列表 (关键字段)api_list:{type:keyword# 用于精确匹配接口是否存在}}}}}}# -----------------# 存入数据格式# -----------------# data {# case_path: am.cases.scp.func.admin.network.topo.dvs.test_dvs.TestDvs#test_tc_topo_business_org_01_001,# case_name: 租户经典网络创建交换机测试,# api_tree: [# {# prefix_step: [{# step_name: admin查询资源池,# details: [# {method: GET, url: /login-info, status: 200},# {method: POST, url: /ticket, status: 200},# {method: GET, url: /admin/azs, status: 200}# ]# }]# },# {# case_step: [# {# step_name: admin创建运营管理员,# details: [# {method: GET, url: /login-info, status: 200},# {method: GET, url: /admin/clusters, status: 200},# {method: GET, url: /admin/azs/{uuid}/overview, status: 200},# {method: POST, url: /admin/msps, status: 200},# {method: GET, url: /admin/ulogs/{uuid}, status: 200}# ]},# {# step_name: admin创建租户,# details: [# {method: GET, url: /login-info, status: 200},# {method: POST, url: /admin/projects, status: 200},# {method: GET, url: /admin/ulogs/{uuid}, status: 200},# ]# },# {# step_name: 登录账号,# details: []# },# ]# },## {# api_list: [ # 存放用例调用的所有接口# GET::/summary, GET::/admin/dhs,# GET::/admin/clusters,# GET::/tenant/ulogs,# GET::/admin/azs/{uuid}/overview,# GET::/admin/network/vpc/topo,# GET::/admin/ulogs/{uuid},# POST::/admin/projects,# ]# }# ]# }defcalculate_str_md5(input_str:str,encoding:strutf-8)-str: 计算字符串的MD5值 :param input_str: 输入字符串 :param encoding: 字符串编码默认utf-8 :return: 32位十六进制MD5字符串 # 创建MD5对象md5_objhashlib.md5()# 更新待哈希的字节数据md5_obj.update(input_str.encode(encoding))# 获取十六进制摘要32位字符串returnmd5_obj.hexdigest()classEsClient():def__init__(self,hostES_HOST,portES_PORT,userES_USER,pwdES_PWD):self.esElasticsearch([{host:host,port:port,scheme:http}],http_auth(user,pwd))self.connect_check()# 验证连接defconnect_check(self):ifself.es.ping():print(✅ ES连接成功集群信息)cluster_infoself.es.info()print(f 集群名称:{cluster_info[cluster_name]})print(f ES版本:{cluster_info[version][number]})print(f 节点名称:{cluster_info[name]})else:print(❌ ES连接失败ping()返回False)# 初始化索引definit_index(self,index_name):ifnotself.es.indices.exists(indexindex_name):self.es.indices.create(indexindex_name,bodymapping_body)else:print(fℹ️ 索引{index_name}已存在跳过创建)# 删除索引defdelete_index(self,index_name):ifself.es.indices.exists(indexindex_name):self.es.indices.delete(indexindex_name)print(f✅ 索引{index_name}删除成功)else:print(fℹ️ 索引{index_name}不存在跳过删除)# 插入数据definsert_data(self,index_name,data):_idcalculate_str_md5(data[case_path])try:# 写入 ESresself.es.index(indexindex_name,id_id,documentdata)print(f 写入成功 ID:{res[_id]}, 结果:{res[result]})exceptExceptionase:print(f❌ 写入失败:{e})# 刷新索引确保写入立即可见self.es.indices.refresh(indexindex_name)# 查询数据defget_data(self,index_name,_id):responseself.es.get(indexindex_name,id_id)print(查询结果:,response[_source])# 通过接口数据查询用例数据defsearch_case_by_api(self,index_name,target_api): 查询 api_list 中包含指定接口的用例路径 # 2. 构建查询 DSL# 使用 term 查询进行精确匹配假设 api_list 是 keyword 类型或数组# 这比 q... 字符串查询更安全不会受特殊字符影响query_body{query:{nested:{path:api_tree,# 指定要遍历的数组字段名query:{term:{api_tree.api_list:target_api# 指定具体的字段路径}}}},_source:[case_path]# 只返回 case_path 字段}try:# 3. 执行搜索responseself.es.search(indexindex_name,bodyquery_body,size10)# 4. 解析结果hitsresponse[hits][hits]totalresponse[hits][total][value]print(f共找到{total}条包含接口 {target_api} 的用例)forhitinhits:# 获取 _source 中的 case_pathcase_pathhit[_source].get(case_path)print(f-{case_path})exceptExceptionase:print(f查询出错:{e})if__name____main__:EsEsClient()# Es.delete_index(index_name)# Es.init_index(index_name)# Es.get_data(index_name, 61da0b051dd70e66bb0789e8ea8b68d5)target_apiGET::/admin/ulogsEs.search_case_by_api(index_name,target_api)