1，ES的儲存結構瞭解

在ES中，儲存結構主要有四種，與傳統的關係型資料庫對比如下：
index（Indices）相當於一個database
type相當於一個table
document相當於一個row
properties（Fields）相當於一個column

Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices -> Types -> Documents -> Fields

2，ES寫入測試

寫入一個文件（一條資料）

PUT http://192.168.1.32:9200/twitter/tweet/377827236
{
"tweet_id": "555555555555555555555666",
"user_screen_name": "kanazawa_mj",
"tweet": "blog3444444",
"user_id": "377827236",
"id": 214019
}

我們看到path:/twitter/tweet/377827236包含三部分資訊：

3，ES查詢測試

查詢一個文件，包含love，返回50條資料，採用展開的json格式

GET http://192.168.1.32:9200/twitter/tweet/_search?q=tweet:love&size=50&pretty=true
{
  "took" : 20,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "failed" : 0
  },
  "hits" : {
    "total" : 11639,
    "max_score"
 
 : 8.448289,
    "hits" : [
      {
        "_index" : "twitter",
        "_type" : "tweet",
        "_id" : "AV0fnFOX6PBTXc6mRjpL",
        "_score" : 8.448289,
        "_source" : {
          "tweet_id" : "843105177913757697",
          "user_screen_name" : "jessicapalapal",
          "tweet" : "Love, love, love ",
          "user_id" : "740434015",
          "id" : 474551
        }
      },
      {
        "_index" : "twitter",
        "_type" : "tweet",
        "_id" : "AV0fni__6PBTXc6mSeyR",
        "_score" : 8.436986,
        "_source" : {
          "tweet_id" : "695096306763583488",
          "user_screen_name" : "SampsonMariel",
          "tweet" : "Love love love^_^ #ALDUB29thWeeksary",
          "user_id" : "2483556636",
          "id" : 723297
        }
      },
      {
        "_index" : "twitter",
        "_type" : "tweet",
        "_id" : "AV0fmxvV6PBTXc6mQ8Mb",
        "_score" : 8.425938,
        "_source" : {
          "tweet_id" : "835676311637086209",
          "user_screen_name" : "thedaveywavey",
          "tweet" : "Love is love is love is love. ",
          "user_id" : "17191297",
          "id" : 311967
        }
      }
    ]
  }
}

4，ES批量寫入測試

• 寫入程式，編寫Python指令碼，生產者和消費者模式，從Mysql資料庫讀取資料，1000條資料寫入一次ES
• 本機環境，Windows，記憶體佔用100M，CPU佔用15%
• ES服務，Ubuntu14.04，CPU佔用5%，記憶體較少
• 單程序，5個寫入執行緒，100萬行資料，500秒
• 單程序，20個寫入執行緒，100萬行資料，500秒
• 補充：據說，修改ES配置，先關閉資料索引，可以提高資料寫入速度，尚未測試

5，下一步計劃

• ES資料分片機制、搜尋引數配置（mapping、filter）等，尚需要根據專案需求，深入學習和測試。
• ES支援的額外功能，例如時間範圍搜尋、中文簡繁體、拼音搜尋、GIS位置搜尋、英文時態支援等。

6，參考資料

7，附件（Python寫入ES程式碼）

# coding=utf-8
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import time
import argparse
import sys
reload(sys)
sys.setdefaultencoding('utf-8')

# ES索引和Type名稱
INDEX_NAME = "twitter"
TYPE_NAME = "tweet"

# ES操作工具類
class es_tool():
    # 類初始化函式
    def __init__(self, hosts, timeout):
        self.es = Elasticsearch(hosts, timeout=5000)
        pass

    # 將資料儲存到es中
    def set_data(self, fields_data=[], index_name=INDEX_NAME, doc_type_name=TYPE_NAME):
        # 建立ACTIONS
        ACTIONS = []
        # print "es set_data length",len(fields_data)
        for fields in fields_data:
            # print "fields", fields
            # print fields[1]
            action = {
                "_index": index_name,
                "_type": doc_type_name,
                "_source": {
                    "id": fields[0],
                    "tweet_id": fields[1],
                    "user_id": fields[2],
                    "user_screen_name": fields[3],
                    "tweet": fields[4]
                }
            }
            ACTIONS.append(action)

        # print "len ACTIONS", len(ACTIONS)
        # 批量處理
        success, _ = bulk(self.es, ACTIONS, index=index_name, raise_on_error=True)
        print('Performed %d actions' % success)

# 讀取引數
def read_args():
    parser = argparse.ArgumentParser(description="Search Elastic Engine")
    parser.add_argument("-i", dest="input_file", action="store", help="input file1", required=False, default="./data.txt")
    # parser.add_argument("-o", dest="output_file", action="store", help="output file", required=True)
    return parser.parse_args()

# 初始化es，設定mapping
def init_es(hosts=[], timeout=5000, index_name=INDEX_NAME, doc_type_name=TYPE_NAME):
    es = Elasticsearch(hosts, timeout=5000)
    my_mapping = {
        TYPE_NAME: {
            "properties": {
                "id": {
                    "type": "string"
                },
                "tweet_id": {
                    "type": "string"
                },
                "user_id": {
                    "type": "string"
                },
                "user_screen_name": {
                    "type": "string"
                },
                "tweet": {
                    "type": "string"
                }
            }
        }
    }
    try:
        # 先銷燬，後建立Index和mapping
        delete_index = es.indices.delete(index=index_name)  # {u'acknowledged': True}
        create_index = es.indices.create(index=index_name)  # {u'acknowledged': True}
        mapping_index = es.indices.put_mapping(index=index_name, doc_type=doc_type_name,
                                                    body=my_mapping)  # {u'acknowledged': True}
        if delete_index["acknowledged"] != True or create_index["acknowledged"] != True or mapping_index["acknowledged"] != True:
            print "Index creation failed..."
    except Exception, e:
        print "set_mapping except", e

# 主函式
if __name__ == '__main__':
    # args = read_args()
    # 初始化es環境
    init_es(hosts=["192.168.1.32:9200"], timeout=5000)
    # 建立es類
    es = es_tool(hosts=["192.168.1.32:9200"], timeout=5000)
    # 執行寫入操作
    tweet_list = [("111","222","333","444","555"), ("11","22","33","44","55")]
    es.set_data(tweet_list)