用 Python 優雅地玩轉 Elasticsearch：實用技巧與最佳實踐
Elasticsearch，這個開源的分布式搜索與數據分析引擎，因其強大的全文搜索功能而廣受歡迎。
盡管Elasticsearch的核心是用Java編寫的，它提供了REST API，讓各種編程語言的開發者都能輕松與之交互，Python當然也不例外。
今天，我們將深入探討如何將 Elasticsearch 與 Python 結合使用，提升我們的項目到新的高度。
1、入門準備
首先，確保我們有一個運行中的 Elasticsearch 8.X 實例、Kibana實例。
圖片
2、Elasticsearch Python 客戶端介紹
在Python項目中，我們可以選擇以下幾個庫與Elasticsearch交互：
elasticsearch-py：官方提供的低級客戶端（Official low-level client for Elasticsearch），直接且靈活。
https://elasticsearch-py.readthedocs.io/en/v8.12.1/
elasticsearch-dsl：基于 elasticsearch-py 的高級封裝，簡化了很多操作，更適合日常使用。
https://elasticsearch-dsl.readthedocs.io/en/latest/
django-elasticsearch-dsl：為 Django 用戶設計，基于elasticsearch-dsl，實現了與Django的深度集成。
https://django-elasticsearch-dsl.readthedocs.io/en/latest/
3、Elasticsearch Python 客戶端適用場景及優缺點
客戶端 | 應用場景 | 優點 | 缺點 |
elasticsearch-py | 直接與Elasticsearch交互的底層操作 | - 完整訪問Elasticsearch API | - 代碼復雜，易出錯 |
elasticsearch-dsl | 構建復雜搜索查詢 | - 簡化查詢構建 | - 學習成本相對較高 |
django-elasticsearch-dsl | 在Django項目中使用Elasticsearch | - 與Django無縫集成 | - 限定于Django項目 |
4、使用 elasticsearch-py 進行增刪改查基礎操作
elasticsearch-py 是Elasticsearch的官方低級Python客戶端。
它允許我們執行所有基本和高級的Elasticsearch操作，包括直接與集群交互、管理索引、執行CRUD（創建、讀取、更新、刪除）操作以及搜索。
以下是使用elasticsearch-py的一些基礎操作示例：
4.1 導入依賴
導入必要的Python庫，包括datetime、Elasticsearch、configparser，并配置警告過濾以忽略警告信息。
from elasticsearch import Elasticsearch
import configparser
import warnings
warnings.filterwarnings("ignore")4.2 初始化Elasticsearch客戶端
init_es_client函數從配置文件config.ini讀取Elasticsearch的配置（如主機地址、用戶名和密碼），并初始化Elasticsearch客戶端。這允許與Elasticsearch集群建立連接。
def init_es_client(config_path='./conf/config.ini'):
"""初始化并返回Elasticsearch客戶端"""
# 初始化配置解析器
config = configparser.ConfigParser()
# 讀取配置文件
config.read(config_path)
# 從配置文件中獲取Elasticsearch配置
es_host = config.get('elasticsearch', 'ES_HOST')
es_user = config.get('elasticsearch', 'ES_USER')
es_password = config.get('elasticsearch', 'ES_PASSWORD')
es = Elasticsearch(
hosts=[es_host],
basic_auth=(es_user, es_password),
verify_certs=False,
ca_certs='conf/http_ca.crt'
)
return es
basic_auth=(es_user, es_password)Elasticsearch 8.X要求客戶端連接時進行身份驗證。這里使用基本認證（HTTP Basic Authentication）提供用戶名和密碼。這兩個值應該對應于有效的Elasticsearch用戶憑證，該用戶需要有足夠的權限執行客戶端請求的操作。
verify_certs=False
這個選項告訴客戶端是否驗證Elasticsearch服務器的TLS證書。在生產環境中，我們應該將其設置為True以確保安全的通信。將此設置為False可能會導致中間人攻擊等安全風險。在開發或測試環境中，如果使用的是自簽名證書，可能需要暫時設置為False來避免驗證錯誤。
ca_certs='conf/http_ca.crt'
當verify_certs=True時，這里指定了CA證書的路徑，客戶端將使用它來驗證服務器證書的簽名。這是實現TLS加密通信的關鍵部分。在Elasticsearch 8.X中，如果啟用了安全特性（默認情況下啟用），那么客戶端需要信任連接到的Elasticsearch服務器使用的CA。如果Elasticsearch使用的是自簽名證書或私有CA簽發的證書，那么我們需要在客戶端提供CA證書的路徑。
對于Elasticsearch 8.X版本，正確配置客戶端以安全地連接到Elasticsearch服務是非常重要的。這包括使用HTTPS協議、提供正確的用戶認證憑證，以及在啟用了TLS加密通信時驗證服務器證書。為了最大化安全性和兼容性，強烈推薦在生產環境中使用由受信任CA簽發的證書，并且始終驗證服務器證書。
4.3 創建索引
create_index函數嘗試創建一個新索引。如果指定的索引名已存在，則忽略創建操作。索引是數據存儲和搜索的基本單位。
def create_index(es, index_name="test-index"):
"""創建索引，如果索引已存在則忽略"""
if not es.indices.exists(index=index_name):
es.indices.create(index=index_name)4.4 定義映射
define_mapping函數為索引設置映射。映射定義了索引中文檔的字段類型，如文本、整數和關鍵詞等。這有助于Elasticsearch理解字段內容并優化搜索和聚合操作。
def define_mapping(es, index_name="test-index"):
"""為索引定義映射"""
mapping = {
"mappings": {
"properties": {
"name": {"type": "text"},
"age": {"type": "integer"},
"email": {"type": "keyword"}
}
}
}
es.indices.create(index=index_name, body=mapping, ignore=400) # ignore=400忽略索引已存在錯誤4.5 插入文檔
insert_document函數向指定索引插入（或更新）一個文檔。文檔由一個Python字典表示，可以包含多個字段和值。如果提供了doc_id，該ID將用于文檔；否則，Elasticsearch會自動生成一個ID。
def insert_document(es, index_name="test-index", doc_id=None, document=None):
"""插入文檔到指定索引"""
es.index(index=index_name, id=doc_id, document=document)4.6 更新文檔
update_document函數更新指定索引中的特定文檔。需要文檔的ID和要更新的字段。
def update_document(es, index_name="test-index", doc_id=None, updated_doc=None):
"""更新指定ID的文檔"""
es.update(index=index_name, id=doc_id, body={"doc": updated_doc})4.7 刪除文檔
delete_document函數從指定索引中刪除特定ID的文檔。
def delete_document(es, index_name="test-index", doc_id=None):
"""刪除指定ID的文檔"""
es.delete(index=index_name, id=doc_id)4.8 搜索文檔
search_documents 函數在指定索引中執行搜索查詢，并返回匹配的文檔。查詢通過一個查詢DSL（Domain-Specific Language）構建，可以非常靈活地定義搜索條件。
def search_documents(es, index_name="test-index", query=None):
"""在指定索引中搜索文檔"""
return es.search(index=index_name, body=query)4.9 main函數
main函數是程序的入口點，按順序執行了創建索引、定義映射、插入文檔、更新文檔、搜索文檔和刪除文檔的操作，演示了與Elasticsearch交互的完整流程。
def main():
# 初始化Elasticsearch客戶端
es = init_es_client()
# 創建索引
create_index(es)
# 定義映射
define_mapping(es)
# 插入文檔
doc = {
"name": "John Doe",
"age": 30,
"email": "john.doe@example.com"
}
insert_document(es, doc_id="1", document=doc)
# 更新文檔
# 注意：這里假設我們知道文檔的ID。實際使用時可能需要通過搜索等方式來確定ID
update_document(es, doc_id="1", updated_doc={"age": 31})
# 搜索文檔
query = {
"query": {
"match": {
"name": "John Doe"
}
}
}
search_result = search_documents( es, query=query )
print( search_result )
# 刪除文檔
delete_document(es, doc_id="1")以上示例展示了使用elasticsearch-py進行基本的Elasticsearch操作。
這些操作涵蓋了創建和刪除索引、定義映射、插入、更新和刪除文檔以及基本的搜索功能。
elasticsearch-py提供了訪問Elasticsearch強大功能的直接途徑，但正如之前討論的，使用它需要對Elasticsearch的工作原理有深入理解。
5、使用 elasticsearch-dsl 進行基礎操作
如下代碼演示了如何使用elasticsearch-dsl，一個Python庫，以便與Elasticsearch進行高效交互。
我們將涵蓋初始化客戶端、創建索引、文檔的CRUD操作以及執行搜索查詢。
5.1 初始化 Elasticsearch 客戶端
為了與Elasticsearch集群交互，首先需要建立連接。我們通過讀取配置文件來獲取連接信息，并創建一個默認連接。
def init_es_client_dsl(config_path='./conf/config.ini'):
config = configparser.ConfigParser()
config.read(config_path)
es_host = config.get('elasticsearch', 'ES_HOST')
es_user = config.get('elasticsearch', 'ES_USER')
es_password = config.get('elasticsearch', 'ES_PASSWORD')
connections.create_connection(
hosts=[es_host],
http_auth=(es_user, es_password),
verify_certs=False
)5.2 創建索引
在Elasticsearch中，索引是存儲文檔的容器。我們定義了一個文檔類 MyDocument，指定了索引名稱和映射，并刪除已存在的同名索引后重新創建。
class MyDocument(Document):
name = Text()
age = Integer()
email = Text()
class Index:
name = 'test-index'
settings = { "number_of_shards": 1, }
def create_index_dsl():
es = connections.get_connection()
es.indices.delete(index='test-index', ignore=[400, 404])
MyDocument.init()5.3 插入文檔
將一個新文檔插入到Elasticsearch。如果提供了id，將使用它作為文檔ID；否則，Elasticsearch會自動生成一個。
def insert_document_dsl(document):
doc = MyDocument(meta={'id': document.get('id', None)}, **document)
doc.save()5.4 更新文檔
根據文檔ID更新已存在的文檔。這里我們更新了文檔的某些字段。
def update_document_dsl(doc_id, updated_doc):
doc = MyDocument.get(id=doc_id)
for key, value in updated_doc.items():
setattr(doc, key, value)
doc.save()5.5 刪除文檔
根據ID刪除指定的文檔。
def delete_document_dsl(doc_id):
doc = MyDocument.get(id=doc_id)
doc.delete()5.6 搜索文檔
執行一個搜索查詢，返回匹配指定查詢條件的文檔。在此例中，我們使用match查詢匹配名字字段。
def search_documents_dsl(query):
es = connections.get_connection()
es.indices.refresh(index="test-index")
s = Search(index="test-index").query("match", name=query)
response = s.execute()
return response5.7 主函數
main_ds l函數串聯了上述所有步驟，展示了如何在實際應用中使用這些功能。
def main_dsl():
init_es_client_dsl()
create_index_dsl()
insert_document_dsl({ ... })
results = search_documents_dsl('John Doe')
update_document_dsl('1', { ... })
delete_document_dsl('1')5.8 運行
將上述代碼保存為Python文件并執行，可以看到從插入到搜索、更新和刪除文檔的完整流程。
圖片
5.9 特別強調——默認連接池管理
在elasticsearch-dsl中，當我們創建查詢或者執行任何需要與Elasticsearch服務器通信的操作時，并不需要每次都顯式地指定Elasticsearch連接實例。
這是因為elasticsearch-dsl內部維護了一個默認的連接池。當我們首次使用connections.create_connection函數創建連接時，如果不指定別名，這個連接就被設置為默認連接。
官方alias 示例：
from elasticsearch_dsl import connections
connections.create_connection(alias='my_new_connection', hosts=['localhost'], timeout=60)后續的所有操作，如搜索查詢，都會自動使用這個默認連接，除非咱們通過using參數顯式指定了另一個連接。
這種設計使得在大多數情況下，我們只需在應用啟動時建立一次連接，而不需要在每個查詢中重復指定連接信息，從而簡化了代碼并提高了代碼的可讀性和維護性。
參見：
https://elasticsearch-dsl.readthedocs.io/en/latest/configuration.html#default-connection
6、小結
篇幅原因，django-elasticsearch-dsl API 沒有展開。如果需要，歡迎留言討論。
在本文中，我們探討了如何將Elasticsearch與Python結合使用，通過兩種主要的Python客戶端——elasticsearch-py和elasticsearch-dsl。
elasticsearch-py提供了直接且靈活的底層API訪問，適用于需要完整控制Elasticsearch交互細節的場景。
相比之下，elasticsearch-dsl提供了更高級的抽象，通過更為Pythonic的接口簡化了復雜搜索查詢的構建，使得代碼更加簡潔易讀，尤其適合日常使用和復雜查詢構建。
此外，我們還介紹了如何通過elasticsearch-dsl內部管理的默認連接池來簡化連接管理，避免了在每次查詢時重復指定連接信息，從而提高了開發效率和代碼的可維護性。通過這種方式，開發者只需在應用啟動時配置一次連接，之后便可以在整個應用中復用這個默認連接。
無論是直接使用elasticsearch-py進行底層操作，還是利用elasticsearch-dsl進行更加高效的數據處理和搜索，Elasticsearch都能為Python開發者提供強大的搜索和數據分析能力，幫助他們輕松應對各種數據處理和搜索需求，將項目提升到新的高度。在選擇合適的客戶端和API時，重要的是根據項目的具體需求和團隊的熟悉程度來做出決策，以確保既能充分發揮Elasticsearch的強大功能，又能保持代碼的可讀性和可維護性。
