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

成都創(chuàng)新互聯(lián)主營壽陽網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營網(wǎng)站建設(shè)方案,手機APP定制開發(fā),壽陽h5成都小程序開發(fā)搭建,壽陽網(wǎng)站營銷推廣歡迎壽陽等地區(qū)企業(yè)咨詢

坦白說，這是第一個python程序，雖然看起來寫的很爛，但是你放心，我試過了無毒，而且運行結(jié)果，既然是正確的！

導(dǎo)出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")

導(dǎo)入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)

到此，關(guān)于“elasticsearch怎么實現(xiàn)導(dǎo)入導(dǎo)出CSV”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
文章標(biāo)題：elasticsearch怎么實現(xiàn)導(dǎo)入導(dǎo)出CSV
路徑分享：http://weahome.cn/article/iepdhi.html