python elasticsearch环境搭建详解
windows下载zip
linux下载tar
下载地址：https://www.elastic.co/downloads/elasticsearch
解压后运行：bin/elasticsearch(orbin\elasticsearch.batonWindows)
检查是否成功：访问http://localhost:9200
linux下不能以root用户运行，
普通用户运行报错：
java.nio.file.AccessDeniedException
原因：当前用户没有执行权限
解决方法：chownlinux用户名elasticsearch安装目录-R
例如：chownealsticsearch/data/wwwroot/elasticsearch-6.2.4-R
PS：其他Java软件报.AccessDeniedException错误也可以同样方式解决，给执行用户相应的目录权限即可
2|0代码实例
如下的代码实现类似链家网小区搜索功能。
从文件读取小区及地址信息写入es,然后通过小区所在城市code及搜索关键字匹配到对应小区。
代码主要包含三部分内容：
1.创建索引
2.用bulk将批量数据存储到es
3.数据搜索
注意：
代码的es版本交低2.xx版本，高版本在创建的索引数据类型有所不同
#coding:utf8
from__future__importunicode_literals
importos
importtime
importconfig
fromdatetimeimportdatetime
fromelasticsearchimportElasticsearch
fromelasticsearch.helpersimportbulk
classElasticSearch():
def__init__(self,index_name,index_type,ip="127.0.0.1"):
'''
:paramindex_name:索引名称
:paramindex_type:索引类型
'''
self.index_name=index_name
self.index_type=index_type
#无用户名密码状态
#self.es=Elasticsearch([ip])
#用户名密码状态
self.es=Elasticsearch([ip],http_auth=('elastic','password'),port=9200)
defcreate_index(self,index_name="ftech360",index_type="community"):
'''
创建索引,创建索引名称为ott，类型为ott_type的索引
:paramex:Elasticsearch对象
:return:
'''
#创建映射
_index_mappings={
"mappings":{
self.index_type:{
"properties":{
"city_code":{
"type":"string",
#"index":"not_analyzed"
},
"name":{
"type":"string",
#"index":"not_analyzed"
},
"address":{
"type":"string",
#"index":"not_analyzed"
}
}
}
}
}
ifself.es.indices.exists(index=self.index_name)isTrue:
self.es.indices.delete(index=self.index_name)
res=self.es.indices.create(index=self.index_name,body=_index_mappings)
printres
defbuild_data_dict(self):
name_dict={}
withopen(os.path.join(config.datamining_dir,'data_output','house_community.dat'))asf:
forlineinf:
line_list=line.decode('utf-8').split('\t')
community_code=line_list[6]
name=line_list[7]
city_code=line_list[0]
name_dict[community_code]=(name,city_code)
address_dict={}
withopen(os.path.join(config.datamining_dir,'data_output','house_community_detail.dat'))asf:
forlineinf:
line_list=line.decode('utf-8').split('\t')
community_code=line_list[6]
address=line_list[10]
address_dict[community_code]=address
returnname_dict,address_dict
defbulk_index_data(self,name_dict,address_dict):
'''
用bulk将批量数据存储到es
:return:
'''
list_data=[]
forcommunity_code,datainname_dict.items():
tmp={}
tmp['code']=community_code
tmp['name']=data[0]
tmp['city_code']=data[1]
ifcommunity_codeinaddress_dict:
tmp['address']=address_dict[community_code]
else:
tmp['address']=''
list_data.append(tmp)
ACTIONS=[]
forlineinlist_data:
action={
"_index":self.index_name,
"_type":self.index_type,
"_id":line['code'],#_id小区code
"_source":{
"city_code":line['city_code'],
"name":line['name'],
"address":line['address']
}
}
ACTIONS.append(action)
#批量处理
success,_=bulk(self.es,ACTIONS,index=self.index_name,raise_on_error=True)
#单条写入单条写入速度很慢
#self.es.index(index=self.index_name,doc_type="doc_type_test",body=action)
print('Performed%dactions'%success)
defdelete_index_data(self,id):
'''
删除索引中的一条
:paramid:
:return:
'''
res=self.es.delete(index=self.index_name,doc_type=self.index_type,id=id)
printres
defget_data_id(self,id):
res=self.es.get(index=self.index_name,doc_type=self.index_type,id=id)
##输出查询到的结果
printres['_source']['city_code'],res['_id'],res['_source']['name'],res['_source']['address']
defget_data_by_body(self,name,city_code):
#doc={'query':{'match_all':{}}}
doc={
"query":{
"bool":{
"filter":{
"term":{
"city_code":city_code
}
},
"must":{
"multi_match":{
"query":name,
"type":"phrase_prefix",
"fields":['name^3','address'],
"slop":1,
}
}
}
}
}
_searched=self.es.search(index=self.index_name,doc_type=self.index_type,body=doc)
data=_searched['hits']['hits']
returndata
if__name__=='__main__':
#数据插入es
obj=ElasticSearch("ftech360","community")
obj.create_index()
name_dict,address_dict=obj.build_data_dict()
obj.bulk_index_data(name_dict,address_dict)
#从es读取数据
obj2=ElasticSearch("ftech360","community")
obj2.get_data_by_body(u'保利','510100')
以上就是全部知识点内容，感谢大家的阅读和对毛票票的支持。