使用Python操作Elasticsearch数据索引的教程

>> bin/elasticsearch -f

>> pip install elasticsearch

from datetime import datetime
from elasticsearch import Elasticsearch
es = Elasticsearch() #create a localhost server connection, or Elasticsearch("ip")
es.create(index="test-index", doc_type="test-type", id=1,
  body={"any":"data", "timestamp": datetime.now()})

from datetime import datetime
from elasticsearch import Elasticsearch
from elasticsearch import helpers
es = Elasticsearch("10.18.13.3")
j = 0
count = int(df[0].count())
actions = []
while (j < count):
   action = {
        "_index": "tickets-index",
        "_type": "tickets",
        "_id": j + 1,
        "_source": {
              "crawaldate":df[0][j],
              "flight":df[1][j],
              "price":float(df[2][j]),
              "discount":float(df[3][j]),
              "date":df[4][j],
              "takeoff":df[5][j],
              "land":df[6][j],
              "source":df[7][j],
              "timestamp": datetime.now()}
        }
  actions.append(action)
  j += 1

  if (len(actions) == 500000):
    helpers.bulk(es, actions)
    del actions[0:len(actions)]

if (len(actions) > 0):
  helpers.bulk(es, actions)
  del actions[0:len(actions)]

#helpers.py source code
def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,
    expand_action_callback=expand_action, **kwargs):
  actions = map(expand_action_callback, actions)

  # if raise on error is set, we need to collect errors per chunk before raising them
  errors = []

  while True:
    chunk = islice(actions, chunk_size)
    bulk_actions = []
    for action, data in chunk:
      bulk_actions.append(action)
      if data is not None:
        bulk_actions.append(data)

    if not bulk_actions:
      return

def bulk(client, actions, stats_only=False, **kwargs):
  success, failed = 0, 0

  # list of errors to be collected is not stats_only
  errors = []

  for ok, item in streaming_bulk(client, actions, **kwargs):
    # go through request-reponse pairs and detect failures
    if not ok:
      if not stats_only:
        errors.append(item)
      failed += 1
    else:
      success += 1

  return success, failed if stats_only else errors

{
  '_op_type': 'delete',
  '_index': 'index-name',
  '_type': 'document',
  '_id': 42,
}
{
  '_op_type': 'update',
  '_index': 'index-name',
  '_type': 'document',
  '_id': 42,
  'doc': {'question': 'The life, universe and everything.'}
}