Я должен хранить некоторые сообщения в ElasticSearch, интегрировать с моей программой на Python. Теперь я пытаюсь сохранить сообщение:
d={"message":"this is message"}
for index_nr in range(1,5):
ElasticSearchAPI.addToIndex(index_nr, d)
print d
Это означает, что если у меня есть 10 сообщений, я должен повторить свой код 10 раз. Итак, я хочу сделать файл сценария или пакетный файл. Я проверил ElasticSearch Guide, BULK API можно использовать. Формат должен быть примерно таким:
{ "index" : { "_index" : "test", "_type" : "type1", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_type" : "type1", "_id" : "2" } }
{ "create" : { "_index" : "test", "_type" : "type1", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_type" : "type1", "_index" : "index1"} }
{ "doc" : {"field2" : "value2"} }
что я сделал, это:
{"index":{"_index":"test1","_type":"message","_id":"1"}}
{"message":"it is red"}
{"index":{"_index":"test2","_type":"message","_id":"2"}}
{"message":"it is green"}
Я также использую инструмент curl для хранения документов.
$ curl -s -XPOST localhost:9200/_bulk --data-binary @message.json
Теперь я хочу использовать свой код Python для сохранения файла в Elastic Search.
from datetime import datetime
from elasticsearch import Elasticsearch
from elasticsearch import helpers
es = Elasticsearch()
actions = [
{
"_index": "tickets-index",
"_type": "tickets",
"_id": j,
"_source": {
"any":"data" + str(j),
"timestamp": datetime.now()}
}
for j in range(0, 10)
]
helpers.bulk(es, actions)
Хотя код @justinachen помог мне начать с py -asticsearch, после просмотра исходного кода позвольте мне сделать простое улучшение:
es = Elasticsearch()
j = 0
actions = []
while (j <= 10):
action = {
"_index": "tickets-index",
"_type": "tickets",
"_id": j,
"_source": {
"any":"data" + str(j),
"timestamp": datetime.now()
}
}
actions.append(action)
j += 1
helpers.bulk(es, actions)
helpers.bulk() уже делает сегментацию за вас. Под сегментацией я подразумеваю чаки, отправляемые каждый раз на сервер. Если вы хотите уменьшить порцию отправленных документов, сделайте: helpers.bulk(es, actions, chunk_size=100)
Немного полезной информации для начала:
helpers.bulk() - это просто оболочка helpers.streaming_bulk но первая принимает список, что делает его удобным.
helpers.streaming_bulk был основан на Elasticsearch.bulk() поэтому вам не нужно беспокоиться о том, что выбрать.
Так что в большинстве случаев helpers.bulk() должен быть всем, что вам нужно.
(другие подходы, упомянутые в этом потоке, используют список Python для обновления ES, что не является хорошим решением сегодня, особенно когда вам нужно добавить миллионы данных в ES)
Лучше всего использовать генераторы Python - обрабатывать гигабайты данных, не выходя из памяти или сильно снижая скорость.
Ниже приведен пример фрагмента из примера практического использования - добавление данных из файла журнала nginx в ES для анализа.
def decode_nginx_log(_nginx_fd):
for each_line in _nginx_fd:
# Filter out the below from each log line
remote_addr = ...
timestamp = ...
...
# Index for elasticsearch. Typically timestamp.
idx = ...
es_fields_keys = ('remote_addr', 'timestamp', 'url', 'status')
es_fields_vals = (remote_addr, timestamp, url, status)
# We return a dict holding values from each line
es_nginx_d = dict(zip(es_fields_keys, es_fields_vals))
# Return the row on each iteration
yield idx, es_nginx_d # <- Note the usage of 'yield'
def es_add_bulk(nginx_file):
# The nginx file can be gzip or just text. Open it appropriately.
...
es = Elasticsearch(hosts = [{'host': 'localhost', 'port': 9200}])
# NOTE the (...) round brackets. This is for a generator.
k = ({
"_index": "nginx",
"_type" : "logs",
"_id" : idx,
"_source": es_nginx_d,
} for idx, es_nginx_d in decode_nginx_log(_nginx_fd))
helpers.bulk(es, k)
# Now, just run it.
es_add_bulk('./nginx.1.log.gz')
Этот скелет демонстрирует использование генераторов. Вы можете использовать это даже на голой машине, если вам нужно. И вы можете продолжать расширять это, чтобы приспособиться к вашим потребностям быстро.
Ссылка Python Elasticsearch здесь.
На данный момент я могу придумать два варианта:
1. Определите имя индекса и тип документа для каждого объекта:
es_client = Elasticsearch()
body = []
for entry in entries:
body.append({'index': {'_index': index, '_type': 'doc', '_id': entry['id']}})
body.append(entry)
response = es_client.bulk(body=body)
2. Укажите индекс и тип документа по умолчанию с помощью метода:
es_client = Elasticsearch()
body = []
for entry in entries:
body.append({'index': {'_id': entry['id']}})
body.append(entry)
response = es_client.bulk(index='my_index', doc_type='doc', body=body)
Работает с:
Версия ES: 6.4.0
ES python lib: 6.3.1