共計 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 小編會繼續努力為大家帶來更多實用的文章！