使用Python操作Elasticsearch數(shù)據(jù)索引的教程

Elasticsearch是一個(gè)分布式、Restful的搜索及分析服務(wù)器，Apache Solr一樣，它也是基于Lucence的索引服務(wù)器，但我認(rèn)為Elasticsearch對(duì)比Solr的優(yōu)點(diǎn)在于：
        輕量級(jí)：安裝啟動(dòng)方便，下載文件之后一條命令就可以啟動(dòng)；
        Schema free：可以向服務(wù)器提交任意結(jié)構(gòu)的JSON對(duì)象，Solr中使用schema.xml指定了索引結(jié)構(gòu)；
        多索引文件支持：使用不同的index參數(shù)就能創(chuàng)建另一個(gè)索引文件，Solr中需要另行配置；
        分布式：Solr Cloud的配置比較復(fù)雜。
環(huán)境搭建
啟動(dòng)Elasticsearch，訪問(wèn)端口在9200，通過(guò)瀏覽器可以查看到返回的JSON數(shù)據(jù)，Elasticsearch提交和返回的數(shù)據(jù)格式都是JSON.    
>> bin/elasticsearch -f

安裝官方提供的Python API，在OS X上安裝后出現(xiàn)一些Python運(yùn)行錯(cuò)誤，是因?yàn)閟etuptools版本太舊引起的，刪除重裝后恢復(fù)正常。    
>> pip install elasticsearch

索引操作

對(duì)于單條索引，可以調(diào)用create或index方法。    
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()})

Elasticsearch批量索引的命令是bulk，目前Python API的文檔示例較少，花了不少時(shí)間閱讀源代碼才弄清楚批量索引的提交格式。
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)]

在這里發(fā)現(xiàn)Python API序列化JSON時(shí)對(duì)數(shù)據(jù)類型支撐比較有限，原始數(shù)據(jù)使用的NumPy.Int32必須轉(zhuǎn)換為int才能索引。此外，現(xiàn)在的bulk操作默認(rèn)是每次提交500條數(shù)據(jù)，我修改為5000甚至50000進(jìn)行測(cè)試，會(huì)有索引不成功的情況。
    
#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

對(duì)于索引的批量刪除和更新操作，對(duì)應(yīng)的文檔格式如下，更新文檔中的doc節(jié)點(diǎn)是必須的。    
{
  '_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.'}
}

常見錯(cuò)誤
        SerializationError：JSON數(shù)據(jù)序列化出錯(cuò)，通常是因?yàn)椴恢С帜硞€(gè)節(jié)點(diǎn)值的數(shù)據(jù)類型
        RequestError：提交數(shù)據(jù)格式不正確
        ConflictError：索引ID沖突
        TransportError：連接無(wú)法建立
性能

上面是使用MongoDB和Elasticsearch存儲(chǔ)相同數(shù)據(jù)的對(duì)比，雖然服務(wù)器和操作方式都不完全相同，但可以看出數(shù)據(jù)庫(kù)對(duì)批量寫入還是比索引服務(wù)器更具備優(yōu)勢(shì)。
Elasticsearch的索引文件是自動(dòng)分塊，達(dá)到千萬(wàn)級(jí)數(shù)據(jù)對(duì)寫入速度也沒(méi)有影響。但在達(dá)到磁盤空間上限時(shí)，Elasticsearch出現(xiàn)了文件合并錯(cuò)誤，并且大量丟失數(shù)據(jù)（共丟了100多萬(wàn)條），停止客戶端寫入后，服務(wù)器也無(wú)法自動(dòng)恢復(fù)，必須手動(dòng)停止。在生產(chǎn)環(huán)境中這點(diǎn)比較致命，尤其是使用非Java客戶端，似乎無(wú)法在客戶端獲取到服務(wù)端的Java異常，這使得程序員必須很小心地處理服務(wù)端的返回信息。