elasticsearch與mongodb分布式集群環境下數據同步

1.ElasticSearch是什么

     ElasticSearch 是一個基於Lucene構建的開源、分布式，RESTful搜索引擎。它的服務是為具有數據庫和Web前端的應用程序提供附加的組件（即可搜索的存儲庫）。ElasticSearch為應用程序提供搜索算法和相關的基礎架構，用戶只需要將應用程序中的數據上載到ElasticSearch數據存儲中，就可以通過RESTful URL與其交互。ElasticSearch的架構明顯不同於它之前的其他搜索引擎架構，因為它是通過水平伸縮的方式來構建的。不同於Solr，它在設計之初的目標就是構建分布式平台，這使得它能夠和雲技術以及大數據技術的崛起完美吻合。ElasticSearch構建在更穩定的開源搜索引擎Lucene之上，它的工作方式與無模式的JSON文檔數據非常類似。

 

 ElasticSearch的關鍵特征

 

•   RESTful風格

 

在所有的ElasticSearch的介紹中都不可避免的提到了它是一種具有RESTful特點的搜索引擎。那么什么是RESTful呢？REST（Representational State Transfer表述性狀態轉移）是一種針對網絡應用的設計和開發方式，可以降低開發的復雜性並提高系統的可伸縮性。REST有一些設計概念和准則，凡是遵循這些准則所開發的應用即具備RESTful風格。在REST風格結構中，所有的請求都必須在一個由URL制定的具體地址的對象上進行。例如，如果用/schools/代表一系列學校的話，/schools/1就代表id為1的那所學校，依次類推。這種設計風格為用戶提供了一種簡單便捷的操作方式，用戶可以通過curl等RESTful API與ElasticSearch進行交互，避免了管理XML配置文件的麻煩。下面將簡單介紹

一下通過curl工具對ElasticSearch進行CRUD(增刪改查)操作。

l  索引構建

為了對一個JSON對象進行索引創建，需要向REST API提交PUT請求，在請求中指定由索引名稱，type名稱和ID組成的URL。即

http://localhost:9200/<index>/<type>/[<id>]

例如：curl -XPUT "http://localhost:9200/movies/movie/1" -d'

{

   "title": "The Godfather",

   "director": "Francis Ford Coppola",

    "year":1972

}'

l  通過ID獲得索引數據

向已經構建的索引發送GET請求，即http://localhost:9200/<index>/<type>/<id>

例如：curl -XGET "http://localhost:9200/movies/movie/1" -d''

后面不帶參數時 -d''不要也可以

l  刪除文檔

通過ID指定的索引刪除單個文檔。URL和索引創建、獲取時相同。

例如：curl -XDELETE "http://localhost:9200/movies/movie/1" -d''

 

 

•   ElasticSearch采用Gateway的概念，使得全備份變得更簡單。

 

由於ElasticSearch是專門為分布式環境設計的，所以怎么去對所有節點的索引信息進行持久化是個問題。當然，除了索引信息以外，還有集群信息，mapping和事務日志等都需要進行持久化。當你的節點出現故障或者集群重啟的時候，這些信息就變得非常重要。ElasticSearch中有一個專門的gateway模塊負責元信息的持久化存儲。（Solr里邊是不是通過Zookeeper在管理這部分？）

 

• ElasticSearch支持facetting(facetedsearch,分面搜索)和precolating

 

分面是指事物的多維度屬性。例如一本書包含主題、作者、年代等方面。而分面搜索是指通過事物的這些屬性不斷篩選、過濾搜索結果的方法。當然這點在Lucene中已經得到了實現，所以Solr也支持faceted searching。至於precolating特性則是ElasticSearch設計中的一大亮點。Precolator(過濾器)允許你在ElasticSearch中執行與上文檔、建立索引、執行查詢這樣的常規操作恰恰相反的過程。通過Precolate API,可以在索引上注冊許多查詢，然后向指定的文檔發送prelocate請求，返回匹配該文檔的注冊查詢。舉個簡單的例子，假設我們想獲取所有包含了”elasticsearch”這個詞的tweet，則可以在索引上注冊一個query語句，在每一條tweet上過濾用戶注冊的查詢，可以獲得匹配每條tweet的那些查詢。下面是一個簡單的示例:

首先，建立一個索引：

curl –XPUT localhost:9200/test

接着，注冊一個對test索引的precolator 查詢，制定的名稱為kuku

---該處在本機測試不成功，還沒找到原因---

curl –XPUT localhost:9200/_precolator/test/kuku –d’{

“query”:{

         “term”:{

                   “field1”:”value1”

         }

}

}’

現在，可以過濾一個文本看看哪些查詢跟它是匹配的

crul –XGETlocalhost:9200/test/type/_precolate –d’{

“doc”:{

         “filed1”:”value1”

}

}’

得到的返回結構如下

{“ok”: true, “matches”: [“kuku”]}

--end--

 

• ElasticSearch的分布式特點

 

ElasticSearch不同於Solr，從設計之初就是面向分布式的應用環境，因此具備很多便於搭建分布式應用的特點。例如索引可以被划分為多個分片，每個分片可以有多個副本，每一個節點可以持有一個或多個分片，自動實現負載均衡和分片副本的路由。另外，ElasticSearch具有self-contained的特點，不必使用Tomcat等servlet容器。ElasticSearch的集群是自發現、自管理的（通過內置的Zen discovery模塊實現），配置十分簡單，只要在config/elasticsearch.yml中配置相同的cluster.name即可。

 

• 支持多種數據源

 

ElasticSearch有一個叫做river的插件式模塊，可以將外部數據源中的數據導入elasticsearch並在上面建立索引。River在集群上是單例模式的，它被自動分配到一個節點上，當這個節點掛掉后，river會被自動分配到另外的一個節點上。目前支持的數據源包括：Wikipedia, MongoDB, CouchDB, RabbitMQ, RSS, Sofa, JDBC, FileSystem,Dropbox等。River有一些指定的規范，依照這些規范可以開發適合於自己的應用數據的插件。

2 elasticsearch如何建立的數據源連接？

     ElasticSearch通過river建立與各個數據源之間的連接。例如mongodb,這種連接方式多半是以第三方插件的方式，由一些開源貢獻者貢獻出來的插件建立與各種類型的數據管理系統以及MQ等建立river,索引數據的。本文主要研究的是MONGODB與ES的結合，用的是richardwilly98開發的river。

https://github.com/richardwilly98/elasticsearch-river-mongodb

 

3 mongodb 集群環境搭建

詳見：http://blog.csdn.net/huwei2003/article/details/40453159

4 elasticsearch 如何對真正分布式mongodb集群建立river，並且索引數據

 

1.  首先下載並且解壓Elasticsearch

unzip elasticsearch-0.90.5.zip

2 下載並且解壓elasticsearch-servicewrapper-master.zip

unzip elasticsearch-servicewrapper-master.zip

cd elasticsearch-servicewrapper-master
mv service /root/gy/elasticsearch-0.90.5/bin

3 啟動elasticsearch 

sh elasticsearch start

4 下載river插件

./plugin --install com.github.richardwilly98.elasticsearch/elasticsearch-river-mongodb/1.7.1

這里值得一提的是river的版本必須與mongodb 和ElasticSearch匹配，如果不匹配，那么river的時候不能將mongodb里面所有的數據index進入es。

匹配規則請見下方：

本次測試用的是 es 1.1.2 + mongodb 2.4.6

https://github.com/richardwilly98/elasticsearch-river-mongodb

5 建立river

curl -XPUT "http://localhost:9200/_river/mongodb/_meta" -d'
{
    "type":"mongodb",
    "mongodb":{
        "servers":[{"host":“192.168.225.131","port":37017}],
        "db":"dbname",
        "collection":"collectionname",
        "gridfs":false,
        "options":{
            "include_fields":["_id","VERSION","ACCESSION","file"]
        }
    },
    "index":{
        "name":"indexname",
        "type":"meta"
    }
}'

注： index 里 name 為索引名 要小寫,type 里的meta 為 collection name

由於本次測試使用的是mongodb sharding 集群環境，所以在river連接時，使用mongos 路由，就能夠正常的把mongo集群中的所有數據都建立索引。

gridfs,options 可不設置

 

#curl 方式建立river （並建立resume索引）
curl -XPUT "localhost:9200/_river/tbJobResume/_meta" -d '
{
"type": "mongodb",
"mongodb": {
    "host": "192.168.225.131",
    "port": "37017",
    "db": "MongoModelJobResume",
    "collection": "tbJobResume"
},
"index": {
    "name": "resume",
    "type": "tbJobResume"} }'

說明：_river/tbJobResume  tbJobResume 我用的是表名，創建每個索引的時候最好不同 -d 后面的 '內容'兩個單引號不要丟了
type 后面是 mongodb 因為用的是 mongodb 數據庫

mongodb: 分別是 ip,port,db(name),collection 就不用解釋了

index: name 要建立的索引名，最好是小寫（應該是必須）

index:type collection名，即該索引對應的數據集合名

驗證：
curl "http://localhost:9200/_river/tbJobResume/_meta"
這樣就建好了resume索引，mongodb如果有數據也會同步過來

特別注意：如果tbJobResume表中有字段是地理坐標，需要map成geo_point類型，在創建索引前設置mapping,如下:

curl -XPUT 'http://localhost:9200/resume' -d '
{
    "mappings": {
    "tbJobResume": {
      "properties": {
        "Location": {
          "type": "geo_point"
        }
      }
    }
  }
}'

設置完后在創建索引

---下面是建的另外一個索引---
curl -XPUT "localhost:9200/_river/tbJobPosition/_meta" -d '
{
"type": "mongodb",
"mongodb": {
    "host": "192.168.225.131",
    "port": "37017",
    "db": "MongoModelJob",
    "collection": "tbJobPosition"
},
"index": {
    "name": "position",
    "type": "tbJobPosition"} }'

curl "http://localhost:9200/_river/tbJobPosition/_meta"

---------------
#curl put索引數據

curl -XPUT "http://localhost:9200/customer/tbCustomer/1" -d'
{
   "_id": 1,
   "Name": "Francis Ford Coppola 1",
    "Sex":1
}'
該方法會創建customer索引並put進一條數據,tbCustomer是type

curl -XPUT 'http://192.168.225.131:9200/dept/employee/32' -d '{ "empname": "emp32"}'
curl -XPUT 'http://192.168.225.131:9200/dept/employee/31' -d '{ "empname": "emp31"}'

該方法也會創建dept索引並put進一條數據,employee是type

創建river並索引的變准模版如下：

$ curl -XPUT "localhost:9200/_river/${es.river.name}/_meta" -d '
{
  "type": "mongodb",
  "mongodb": { 
    "servers":
    [
      { "host": ${mongo.instance1.host}, "port": ${mongo.instance1.port} },
      { "host": ${mongo.instance2.host}, "port": ${mongo.instance2.port} }
    ],
    "options": { 
      "secondary_read_preference" : true, 
      "drop_collection": ${mongo.drop.collection}, 
      "exclude_fields": ${mongo.exclude.fields},
      "include_fields": ${mongo.include.fields},
      "include_collection": ${mongo.include.collection},
      "import_all_collections": ${mongo.import.all.collections},
      "initial_timestamp": {
        "script_type": ${mongo.initial.timestamp.script.type},
        "script": ${mongo.initial.timestamp.script}
      },
      "skip_initial_import" : ${mongo.skip.initial.import},
      "store_statistics" : ${mongo.store.statistics},
    },
    "credentials":
    [
      { "db": "local", "user": ${mongo.local.user}, "password": ${mongo.local.password} },
      { "db": "admin", "user": ${mongo.db.user}, "password": ${mongo.db.password} }
    ],
    "db": ${mongo.db.name}, 
    "collection": ${mongo.collection.name}, 
    "gridfs": ${mongo.is.gridfs.collection},
    "filter": ${mongo.filter}
  }, 
  "index": { 
    "name": ${es.index.name}, 
    "throttle_size": ${es.throttle.size},
    "bulk_size": ${es.bulk.size},
    "type": ${es.type.name}
    "bulk": {
      "actions": ${es.bulk.actions},
      "size": ${es.bulk.size},
      "concurrent_requests": ${es.bulk.concurrent.requests},
      "flush_interval": ${es.bulk.flush.interval}
    }
  }
}'

--template end--

--url--

本插件git地址：https://github.com/laigood/elasticsearch-river-mongodb

 

6 測試例子

連接mongo集群，meta collection數據量有22394792條數據

查看ES數據量

最后我在master1 master2 master3上都建立了ElasticSearch,並且3台es rebalance成功，並且數據的總數任然為22394792.

 

  

• 上一篇Elasticsarch及插件安裝
• 下一篇linux下redis的安裝及配置啟動