Может кто-нибудь посоветует, как использовать функциюasticsearch.helpers.streaming_bulk вместо эластичного поиска.helpers.bulk для индексации данных вasticsearch.

Если я просто изменю streaming_bulk вместо объемного, ничего не будет проиндексировано, поэтому я думаю, что его нужно использовать в другой форме.

Приведенный ниже код создает данные индекса, типа и индекса из файла CSV кусками по 500 элементов в эластичный поиск. Это работает должным образом, но я блуждаю, возможно ли увеличить предварительную производительность. Вот почему я хочу попробовать функцию streaming_bulk.

В настоящее время мне нужно 10 минут, чтобы проиндексировать 1 миллион строк для CSV-документа размером 200 МБ. Я использую две машины, Centos 6.6 с 8 CPU, x86_64, CPU MHz: 2499,902, Mem: 15,574G всего. Не уверен, что это может идти быстрее.

es = elasticsearch.Elasticsearch([{'host': 'uxmachine-test', 'port': 9200}])
index_name = 'new_index'
type_name = 'new_type'
mapping = json.loads(open(config["index_mapping"]).read()) #read mapping from json file

es.indices.create(index_name)
es.indices.put_mapping(index=index_name, doc_type=type_name, body=mapping)

with open(file_to_index, 'rb') as csvfile:
    reader = csv.reader(csvfile)        #read documents for indexing from CSV file, more than million rows
    content = {"_index": index_name, "_type": type_name}
    batch_chunks = []
    iterator = 0

    for row in reader:
        var = transform_row_for_indexing(row,fields, index_name, type_name,id_name,id_increment)
        id_increment = id_increment + 1
        #var = transform_row_for_indexing(row,fields, index_name, type_name)
        batch_chunks.append(var)
        if iterator % 500 == 0:
            helpers.bulk(es,batch_chunks)
            del batch_chunks[:]
            print "ispucalo batch"
        iterator = iterator + 1
    # indexing of last batch_chunk
    if len(batch_chunks) != 0:
        helpers.bulk(es,batch_chunks)