elasticsearch怎么實現導入導出CSV

這篇文章主要介紹“elasticsearch怎么實現導入導出CSV”，在日常操作中，相信很多人在elasticsearch怎么實現導入導出CSV問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”elasticsearch怎么實現導入導出CSV”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

坦白說，這是第一個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”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！