Elastic Stack介紹

https://blog.csdn.net/xb_workspace/article/details/85165070

什么是Elasticsearch

分布式高性能高可用可伸縮的的搜索和分析系統

1.什么是搜索

垂直搜索(站內搜索)
互聯網的搜索:電商網站,招聘網站新聞網站各種APP
IT系統的搜索:OA軟件,辦公自動化軟件,會議管理,日程管理,項目管理,員工管理

★搜索引擎原理

2.如果用數據庫做搜索會怎么樣

(1)每條記錄的指定字段的文本,可能會很長,比如"商品描述"字段的長度,這時如果每次都要對每條記錄的所有文本進行掃描,比較耗時.

(2)還不能將搜索詞拆分開來,比如輸入"海市樓",就搜索不出海市蜃樓,用數據庫來實現搜索,性能較差.

3.什么是全文搜索和Lucene

(1)全文搜索,倒排索引
(2)Lucene就是一個jar包,里面包含了封裝好的各種建立倒排索引,以及進行搜索的代碼,包括各種算法.我們用Java開發的時候,引入Lucene jar,然后基于Lucene開發就可以了.用Lucene可以將已有的數據建立索引,Lucene會在本地磁盤給我們組織索引的數據結構

https://www.cnblogs.com/xiaobai1226/p/7652093.html

安裝Elasticsearch

github搜索elasticsearch-rtf,這是elasticsearch中文發行版，針對中文集成了相關插件，方便新手學習測試.

下載完成后,打開壓縮文件,在cmd中運行bin/elasticsearch.bat

然后在瀏覽器打開http://127.0.0.1:9200/

安裝head和kibana

github搜索elasticsearch-head
head插件類似于管理數據庫的Navicat,基于瀏覽器(一個前端管理界面)

下載完成后解壓,按照如下步驟進行(npm可用cnpm代替)

git clone git://github.com/mobz/elasticsearch-head.git cd elasticsearch-head npm install

啟動

npm run start

安裝完成并啟動后,我們在瀏覽器打開 http://127.0.0.1:9100/ ,但此時會發現顯示未連接,這是由于ES默認安全策略,不允許安裝其他第三方插件,我們需要配置elasticsearch與head互通.

在/config/elasticsearch.yml文件末尾添加一下內容

http.cors.enabled: true http.cors.allow-origin: "*"

重啟es和head,可以看到此時已經正常

kibana版本要和elasticsearch-rtf版本一致,所以我們直接百度搜索kibana5.1.1下載
下載完成后解壓,然后cmd中執行/bin/下的kibana.bat,然后在瀏覽器中打開
http://127.0.0.1:5601/

快速啟動Elasticsearch

Windows下建立如下bat文件

start D:\Python\Elasticsearch\elasticsearch-5.1.1\bin\elasticsearch.bat start D:\Python\Elasticsearch\kibana-5.1.1-windows-x86\bin\kibana.bat@echo off :: nodejs安裝目錄下的nodevars.bat set nodevars = "C:\Program Files\nodejs\nodevars.bat" :: 移動到需要啟動的目錄 cd d:/Python/Elasticsearch/elasticsearch-head-master :: 啟動項目 cmd /c %nodevars%&&npm run start@echo off python D:\Python\ESsearch\manage.py runserver cmd /k

Docker安裝Elasticsearch

Elasticsearch概念

★倒排索引

https://www.cnblogs.com/zlslch/p/6440114.html

基本概念

集群：一個或多個節點組織在一起
節點：一個集群中的一臺服務器
分片：索引劃分為多份的能力，允許水平分割，擴展容量，多個分片響應請求
副本：分片的一份或多分，一個節點失敗，其他節點頂上

Elasticsearch 與 MySQL 概念對比

ElasticsearchMySQL

index(索引)	數據庫
type(類型)	表
documents(文檔)	行
field(字段)	列

創建索引

方法一:瀏覽器kibana界面中dev tools中輸入以下語句,(可以類比SQL語句)

PUT lagou {"settings": {"index":{"number_of_shards":5,"number_of_replicas":1}} }

之后我們可以在es界面看到已創建好的索引

方法二:也可以在es界面,手動點擊創建索引

獲取索引

獲取lagou索引的settings

GET lagou/_settings

獲取所有索引的settings,下面兩條語句作用相同

GET _all/_settings GET _settings

獲取多個索引的settings

GET .kibana,lagou/_settings

獲取索引的信息,上面只是獲取了settings

GET lagou

修改settings

我們可以修改number_of_replicas, 即副本的數量, 但是number_of_shards不能修改

PUT lagou/_settings {"number_of_replicas":2 }

增加信息

• job是type,我們需要定義
• id也可以不定義,系統會自動添加

PUT lagou/job/1 {"title":"Python Cookbook","salary_min":15000,"city":"北京","company":{"name":"baidu","company_addr":"beijing"},"publish_data":"2019-01-01","comments":15 }

我們可以在http://127.0.0.1:9100/中查看我們增加的數據

如果我們只想查看source中部分內容,可以用下面的查詢方式

修改數據

兩種方式

• PUT覆蓋修改
• POST增量修改

第一種方式

PUT lagou/job/1 {"title":"Python Cookbook","salary_min":15000,"city":"北京","company":{"name":"baidu","company_addr":"beijing"},"publish_data":"2019-01-01","comments":10 } GET lagou/job/1

第二種方式

POST lagou/job/1/_update {"doc":{"comments":20} } GET lagou/job/1

刪除

1.刪除文檔

DELETE lagou/job/1

2.不能刪除type

DELETE lagou/job

3.刪除索引

DELETE lagou

批量操作

為什么需要批量操作,因為之前的方式每次都建立HTTP請求,三次握手,比較耗時.
現在我們創建了如下圖的索引以及類型

如果要同時查詢兩個type(表)中的內容

GET _mget {"docs":[{"_index":"testdb","_type":"job1","_id":1},{"_index":"testdb","_type":"job2","_id":2}] }

可以簡寫如下

GET testdb/_mget {"docs":[{"_type":"job1","_id":1},{"_type":"job2","_id":2}] }

查詢job1下的兩個文檔(兩行)

GET testdb/job1/_mget {"docs":[{"_id":1},{"_id":2}] }

或者簡寫成

GET testdb/job1/_mget {"ids":[1,2] }

elasticsearch的bulk批量操作

可以合并多個操作，比如index，delete，update，create等等，包括從一個索引到另一個索引：

action_and_meta_data\n
option_source\n
action_and_meta_data\n
option_source\n
….
注意,每兩行是一條數據(index，update，create)(delete除外)

看下面的例子,利用bulk同時增加兩條數據

POST _bulk {"index":{"_index":"lagou","_type":"job1","_id":"1"}} {"title":"python分布式爬蟲開發","salary_min":15000,"city":"北京","company":{"name":"百度","company_addr":"北京市軟件園"},"public_date":"2019-4-16","comments":15} {"index":{"_index":"lagou","_type":"job2","_id":"2"}} {"title":"python django","salary_min":15000,"city":"成都","company":{"name":"百度","company_addr":"成都市軟件園"},"public_date":"2019-4-1","comments":50}

映射

創建索引的時候,可以預先定義字段的類型以及相關屬性

• 類比數據庫中,我們定義一個表,然后要給每一列定義數據類型(整型,字符型),對于ES來說也是一樣,放數據時,對每一個字段指定一種類型.
• 列類型一旦通過映射創建就不可修改(和關系數據庫的不同)

相關屬性的配置
類型

類型解釋

String類型	分為兩種text、keyword, text會對內部的內容進行分析，索引，進行倒排索引等 keyword則會當成字符串，不會被分析，只能完全匹配才能找到String(在es5已經被廢棄了)
日期類型	date 以及datetime等
數字類型	integer long double等等
bool類型	boolean
binary類型	binary
復雜類型	object、nested(數組形式)
geo類型	geo-point地理位置
專業類型	ip competition
object	json里面內置的還有下層{}的對象

下圖紅框中就是一個nested類型

屬性

具體的使用可以看下下面紅框內

Elasticsearch查詢

elasticserach查詢大概分為三類：

• 基本查詢
• 組合查詢
• 過濾：查詢同時，通過filter條件在不影響打分的情況下篩選數據

match查詢(使用較多)

后面為關鍵詞，關于python的都會提取出來，match查詢會對內容進行分詞，并且會自動對傳入的關鍵詞進行大小寫轉換，內置ik分詞器會進行切分，如python網站，只要搜到存在的任何一部分，都會返回

GET lagou/job/_search {"query":{"match":{"title":"python"}} }

term查詢

區別，對傳入的值不會做任何處理，就像keyword(見上面的映射.類型)，只能查包含整個傳入的內容的，一部分也不行，只能完全匹配(例如下面python爬蟲,只能找到完全一樣的關鍵詞)

GET lagou/job/_search {"query":{"term":{"title":"python爬蟲"}} }

terms查詢

title里傳入多個值，只要有一個匹配，就會返回結果

GET lagou/job/_search {"query":{"term":{"title":["工程師","Django"]}} }

from控制查詢的返回數量

通過這里就可以完成分頁處理，從第一條開始查詢兩條,from從0開始

GET lagou/_serach {"query":{"match":{"title":"python"}},"from":1,"size":2 }

match_all 返回所有

GET lagou/_search { “query”:{ “match_all”:{} } }

match_phrase查詢:短語查詢

“python系統”，將其分詞，分為詞條，滿足詞條里面的所有詞才會返回結果，slop參數說明兩個詞條之間的最小距離,比如python和系統之間的距離

GET lagou/_search {"query":{"match_phrase":{"title":{"query":"python系統","slop":6}}} }

sort將結果排序

先查詢全部,然后對評論數按照降序排序

GET lagou/_search {"query";{"match_all":{}},"sort":[{"comments":{"order":"desc"}}] }

range范圍查詢

range是在query里面的，boost是權重，gte lte是大于等于,小于等于的意思,(帶e是等于)
對時間的范圍查詢，則是以字符串的形式傳入

GET lagou / _search {“query”; {“range”: {“comments”: {“gte”: 10,“lte”: 20,“boost”: 2.0}}} }

★range范圍查詢原理

https://zhuanlan.zhihu.com/p/164538758

用的最多的是match,range!!!

bool組合查詢

bool查詢包括了must、should、must_not filter來完成
格式如下：

bool:{"filter":[],"must":[],"should":[],"must_not":[], } 關鍵字作用

filter	用于過濾字段,不參與打分
must	數組里的所有查詢必須全部滿足
should	與must相反,滿足一個或多個都可以

首先我們先插入數據

簡單的filter查詢

select * from testjob where salary=20

term改為match也可以

select * from testjob where title="Python"

查詢時用小寫python
查看分析器解析的結果


select * from testjob where (salary=20 OR title=“python”) AND (salary!=30)
select * from testjob where title=“python” or (title=“django” AND salary=30)
過濾空、非空

Scrapy整合項目

將scrapy爬到的item保存到es

新建一個models文件夾,新建es-jobbole.py文件,內容如下
這個文件作用是創建了索引,類似于我們新建了一個數據庫,但是還沒有往里面放值.

# _*_ coding: utf-8 _*_from datetime import datetime from elasticsearch_dsl import DocType, Date, Nested, Boolean, \analyzer, Completion, Keyword, Text, Integerfrom elasticsearch_dsl.analysis import CustomAnalyzer as _CustomAnalyzerfrom elasticsearch_dsl.connections import connections# 與服務器進行連接，hosts允許多個,但我們現在連接本地 connections.create_connection(hosts=["localhost"])# class CustomAnalyzer(_CustomAnalyzer): # def get_analysis_definition(self): # return {} # ik_analyzer = CustomAnalyzer("ik_max_word", filter=["lowercase"])class ArticleType(DocType):# 伯樂在線文章類型# suggest = Completion(analyzer=ik_analyzer)title = Text(analyzer="ik_max_word") #需要對title進行分詞create_date = Date()url = Keyword()url_object_id = Keyword()front_image_url = Keyword()front_image_path = Keyword()praise_nums = Integer()comment_nums = Integer()fav_nums = Integer()tags = Text(analyzer="ik_max_word")content = Text(analyzer="ik_max_word")class Meta:index = "jobbole" #相當于數據庫名doc_type = "article" #表名if __name__ == "__main__":ArticleType.init() #根據我們定義的類,直接生成mapping#這樣執行之后,那么是定義的哪一個index,哪個type之下呢,所以我們需要Meta

單獨執行這個文件后,我們可以在es界面看到創建的jobbole索引

現在我們寫pipelines.py使爬到的數據真正保存到es

class ElasticsearchPipeline(object):#將數據寫入esdef process_item(self,item,spider):#將item轉換為es的數據,需要elastcsearch-dsl(python接口包)article = ArticleType()article.title = item['title']article.create_date = item['create_date']article.content = remove_tags(item['content'])article.front_image_url = item['front_image_url']if "front_image_path" in item:article.front_image_path = item['front_image_path']article.praise_nums = item["praise_nums"]article.fav_nums = item['fav_nums']article.comment_nums = item['comment_nums']article.url = item['url']article.tags = item['tags']article.meta.id = item['url_object_id']# article.suggest = [{"input":[],"weight":2}]#10,7代表的是權重article.suggest = gen_suggest(ArticleType._doc_type.index,((article.title,10),(article.tags,7)))article.save()return item

然后在settings.py中更改pipeline的執行順序

ITEM_PIPELINES = {# 'ArticleSpider.pipelines.JsonWithEncodingPipeline': 2,# 'scrapy.pipelines.images.ImagesPipeline':1,# 'ArticleSpider.pipelines.ArticleImagePipeline':1#'ArticleSpider.pipelines.MysqlPipeline':1# 'ArticleSpider.pipelines.MysqlTwistedPipeline':1'ArticleSpider.pipelines.ElasticsearchPipeline':1 }

執行main.py文件后,去 http://127.0.0.1:9100/ 查看

Django搭建搜索界面

文件目錄如下

前端界面文件放在/templates文件夾 , css、js文件放在/static文件夾,為了使前端界面能夠找到css和js , 我們在settings.py中添加

STATIC_URL = '/static/'STATICFILES_DIRS = [os.path.join(BASE_DIR, "static") ]

然后在urls.py中添加

from django.conf.urls import url from django.contrib import admin from django.views.generic import TemplateViewurlpatterns = [url(r'^admin/', admin.site.urls),#url(r'^index', index,name = 'index'),url(r'^$',TemplateView.as_view(template_name="index.html"),name="index")]

進入項目所在目錄,找到manage.py目錄打開cmd,輸入下面命令啟動Django

python manage.py runserver

瀏覽器打開 http://127.0.0.1:8000/ ,可以看到如下界面

為了實現搜索建議功能,首先來看一下編輯距離
編輯距離是一種字符之間相似程度的計算方法:linux、linx,即兩個字符串之間的編輯距離等于使一個字符串變成另外一個字符串而進行的(1)插入、(2)刪除、(3)替換或(4)相鄰字符交換位置而進行操作的最少次數.
比如: ed("recognize","recoginze") == 1, 需要交換兩個相鄰i和n .

編輯距離會在fuzzy搜索 中使用到,下面是一個簡單的fuzzy搜索案例:
fuzziness就是編輯距離
prefix_length是前綴長度

GET /_search {"query": {"fuzzy": {"title": {"value": "linux","fuzziness": 0.5,"prefix_length": 0}}},"_source": ["title"] }

ES的suggest

POST /jobbole/_search?pretty {"suggest": {"my_suggest" : {"text" : "linux","completion" : {"field" : "suggest","fuzzy":{"fuzziness":2}}}},"_source": "title" }

實現搜索建議

什么是搜索建議,就是如下圖所示那樣,當 我們在搜索框內輸入某個關鍵詞時,會自動給出聯想提示

官方API
https://www.elastic.co/guide/en/elasticsearch/reference/current/search-suggesters-completion.html
首先在es_jobbole.py中重寫suggest,完成后代碼如下:

# _*_ coding: utf-8 _*_from datetime import datetime from elasticsearch_dsl import DocType, Date, Nested, Boolean, \analyzer, Completion, Keyword, Text, Integerfrom elasticsearch_dsl.analysis import CustomAnalyzer as _CustomAnalyzerfrom elasticsearch_dsl.connections import connections # 與服務器進行連接，hosts允許多個,但我們現在連接本地 connections.create_connection(hosts=["localhost"])#搜索建議,需要從elasticsearch_dsl.analysis導入CustomAnalyzer class CustomAnalyzer(_CustomAnalyzer):def get_analysis_definition(self):return {} ik_analyzer = CustomAnalyzer("ik_max_word", filter=["lowercase"])class ArticleType(DocType):# 伯樂在線文章類型suggest = Completion(analyzer=ik_analyzer) #這個會有問題,所以我們自己重寫title = Text(analyzer="ik_max_word") #需要對title進行分詞create_date = Date()url = Keyword()url_object_id = Keyword()front_image_url = Keyword()front_image_path = Keyword()praise_nums = Integer()comment_nums = Integer()fav_nums = Integer()tags = Text(analyzer="ik_max_word")content = Text(analyzer="ik_max_word")class Meta:index = "jobbole" #相當于數據庫名doc_type = "article" #表名if __name__ == "__main__":ArticleType.init() #根據我們定義的類,直接生成mapping#這樣執行之后,那么是定義的哪一個index,哪個type之下呢,所以我們需要Meta

添加的代碼如下

from elasticsearch_dsl.analysis import CustomAnalyzer as _CustomAnalyzer#搜索建議,需要從elasticsearch_dsl.analysis導入CustomAnalyzer class CustomAnalyzer(_CustomAnalyzer):def get_analysis_definition(self):return {} ik_analyzer = CustomAnalyzer("ik_max_word", filter=["lowercase"]) ...... ...... ...... suggest = Completion(analyzer=ik_analyzer)

報錯1

TypeError: analyze() got an unexpected keyword argument 'analyzer'

解決1
參考: https://blog.csdn.net/javakklam/article/details/80114837

words = es.indices.analyze(index=index,body={'text':text,'analyzer':"ik_max_word"},params={'filter':["lowercase"]})

報錯2

analyzed_words = set([r['token'] for r in words if len(r['token'])>1]) TypeError: string indices must be integers

解決2

analyzed_words = set([r["token"] for r in words["tokens"] if len(r["token"]) > 1])

生成搜索建議詞
這部分我們在pipelines.py寫

def gen_suggest(index,info_tuple):used_words = set()suggests = []for text,weight in info_tuple:if text:words = es.indices.analyze(index=index,body={'text':text,'analyzer':"ik_max_word"},params={'filter':["lowercase"]})#analyzed_words = set([r['token'] for r in words if len(r['token'])>1])analyzed_words = set([r["token"] for r in words["tokens"] if len(r["token"]) > 1])new_words = analyzed_words - used_words #集合相減,把已經存在的詞過濾掉else:new_words = set()if new_words:suggests.append({"input":list(new_words),"weight":weight})return suggestsclass ElasticsearchPipeline(object):#將數據寫入esdef process_item(self,item,spider):#將item轉換為es的數據,需要elastcsearch-dsl(python接口包)article = ArticleType()article.title = item['title']article.create_date = item['create_date']article.content = remove_tags(item['content'])article.front_image_url = item['front_image_url']if "front_image_path" in item:article.front_image_path = item['front_image_path']article.praise_nums = item["praise_nums"]article.fav_nums = item['fav_nums']article.comment_nums = item['comment_nums']article.url = item['url']article.tags = item['tags']article.meta.id = item['url_object_id']# article.suggest = [{"input":[],"weight":2}]#10,7代表的是權重article.suggest = gen_suggest(ArticleType._doc_type.index,((article.title,10),(article.tags,7)))article.save()return item

處理上面的錯誤后,我們運行main.py,爬到一些數據后停止運行,去es界面查看生成的suggest

報錯1
對于Es搜索自動提示功能debug出現Search object has no attribute execute_suggest.的解決方法

這一問題的出現主要是elasticsearch-dsl的改版，導致以前的方法被新版的方法替代或者不再使用。
舊版本代碼：

class SearchSuggest(View):def get(self, request):key_words = request.GET.get('s', '')re_datas = []if key_words:s = ArticleType.search()s = s.suggest('my_suggest', key_words, completion={"field": "suggest", "fuzzy": {"fuzziness": 2},"size": 10}suggestions = s.execute_suggest()for match in suggestions.my_suggest[0].options:source = match._sourcere_datas.append(source["title"])return HttpResponse(json.dumps(re_datas), content_type="application/json")

新版本代碼：
新版本之后execute_suggest()方法就不再使用，取而代之的是execute()方法

class SearchSuggest(View):def get(self, request):key_words = request.GET.get('s', '')re_datas = []if key_words:s = ArticleType.search()s = s.suggest('my_suggest', key_words, completion={"field": "suggest", "fuzzy": {"fuzziness": 2},"size": 10})suggestions = s.execute() suggestion = suggestions.suggestfor match in suggestion.my_suggest[0].options:source = match._sourcere_datas.append(source["title"])return HttpResponse(json.dumps(re_datas), content_type="application/json")

報錯2

AttributeError: 'Response' object has no attribute 'index'

原因是解析json返回的數據時出錯,應改為
for match in suggestions.suggest.my_suggest[0].options[:10]:

搜索展示界面

報錯

HTTPConnectionPool(host='localhost', port=9200): Read timed out. (read timeout=10))

原因是elasticsearch掛了,重啟一下就好了

---

2019-5-15 后續常用功能

在原有基礎上增加字段(mapping映射)

PUT jingdong/comment/_mapping {"properties": {"wordcloud_negative": {"type": "string"},"proportion_positive": {"type": "double"},"proportion_negative": {"type": "double"}} }

更新某個字段內容或者用GET全部更新

POST jingdong/comment/2/_update {"doc":{"comments":20} }

SpringBoot 整合 ES

官方文檔
https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high.html

準備工作

功能測試(坑)

同步mysql數據到ElasticSearch

Elasticsearch就是專門用于搜索的，所以我們會將新增的數據放入Elasticsearch中用于搜索，就有這樣的常見的場景，需要將Mysql中的數據同步到Elasticsearch中。
Mysql數據同步到ES中分為兩種，分別是全量同步和增量同步。

• 全量同步是指全部將數據同步到es，通常是剛建立es，第一次同步時使用。
• 增量同步是指將后續的更新、插入記錄同步到es。（刪除記錄沒有辦法同步，只能兩邊執行自己的刪除命令）

Mysql與ES的數據同步實現方式, 業界有一些開源方案，開源中間件來實現:

基于Mysql的binlog日志訂閱：binlog日志是Mysql用來記錄數據實時的變化。 這里主要的是binlog同步組件，目前實現的有國內的阿里巴巴開發的canal。

ES官方數據收集和同步組件logstash

總結

以上是生活随笔為你收集整理的Elasticsearch学习记录的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。