189 8069 5689
这篇文章主要介绍“elasticsearch怎么实现导入导出CSV”，在日常操作中，相信很多人在elasticsearch怎么实现导入导出CSV问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”elasticsearch怎么实现导入导出CSV”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
成都创新互联主营寿阳网站建设的网络公司,主营网站建设方案,手机APP定制开发,寿阳h5成都小程序开发搭建,寿阳网站营销推广欢迎寿阳等地区企业咨询
坦白说，这是第一个python程序，虽然看起来写的很烂，但是你放心，我试过了无毒，而且运行结果，既然是正确的！
导出CSV
import csv
import sys
import logging
import datetime
from elasticsearch import Elasticsearch
reload(sys)
sys.setdefaultencoding('gbk')
logging.basicConfig()
es = Elasticsearch()
def exportCSV(indexName):
count = 0
finish=False
csvfile = file(indexName+'.csv','wb')
writer = csv.writer(csvfile)
starttime = datetime.datetime.now()
searchRes = es.search(index=indexName,size=100,body={"query": {"match_all": {}}},search_type="scan",scroll="60s")
while True:
scrollRes=es.scroll(scroll_id=searchRes["_scroll_id"],scroll="60s",ignore=[400, 404])
res_list = scrollRes["hits"]["hits"]
data=[]
if not len(res_list) or finish:
break
if count==0:
writer.writerow(tuple(res_list[0]["_source"].keys()))
for item in res_list:
#print tuple(item["_source"].values())
data.append(tuple(item["_source"].values()))
count+=1
if count>=100000:
finish=True
break
writer.writerows(data)
csvfile.close()
endtime = datetime.datetime.now()
print "export size = "+str(count)
print "export cost = "+str(endtime - starttime)
if __name__=="__main__":
exportCSV("test")导入CSV
# -*- coding:utf-8 -*-
import csv
import sys
import os
import logging
import datetime
from elasticsearch import Elasticsearch
from elasticsearch import helpers
reload(sys)
sys.setdefaultencoding('gbk')
logging.basicConfig()
es = Elasticsearch()
def importCSV(indexName,typeName,fileName):
if not os.path.exists(fileName):
print "file not found"
return
actions=[]
if not es.indices.exists(index=indexName,allow_no_indices=True):
#print "not found index"
es.indices.create(index=indexName,body={},ignore=400)
for item in csv.DictReader(open(fileName, 'rb')):
actions.append({"_index":indexName,"_type":typeName,"_source":encoding(item)})
res = helpers.bulk(es,actions,chunk_size=100)
es.indices.flush(index=[indexName])
return len(actions)
def encoding(item):
for i in item:
item[i]=str(item[i]).encode('utf-8')
return item
if __name__=="__main__":
starttime = datetime.datetime.now()
result=importCSV("test","base","test.csv")
print "import size = "+str(result)
endtime = datetime.datetime.now()
print "import cost = "+str(endtime - starttime)到此，关于“elasticsearch怎么实现导入导出CSV”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！