共计 2360 个字符，预计需要花费 6 分钟才能阅读完成。

这篇文章主要介绍“elasticsearch 怎么实现导入导出 CSV”，在日常操作中，相信很多人在 elasticsearch 怎么实现导入导出 CSV 问题上存在疑惑，丸趣 TV 小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”elasticsearch 怎么实现导入导出 CSV”的疑惑有所帮助！接下来，请跟着丸趣 TV 小编一起来学习吧！

坦白说，这是第一个 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”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注丸趣 TV 网站，丸趣 TV 小编会继续努力为大家带来更多实用的文章！