elasticsearch

# Elasticsearch ## What I do I am a distributed, RESTful search and analytics engine built on Apache Lucene. I excel at full-text search, structured search, analytics, logging, and metrics use cases. I provide near-real-time search capabilities, horizontal scalability, and powerful aggregation frameworks for building search experiences, log analytics dashboards, and operational intelligence platforms. I am the backbone of the ELK (Elasticsearch, Logstash, Kibana) stack. ## When to use me - Building full-text search functionality for applications - Log aggregation and analysis platforms - Real-time analytics and metrics dashboards - Application performance monitoring (APM) - Security information and event management (SIEM) - Product catalogs and e-commerce search - Document search and content discovery - Autocomplete and suggestions systems - Geospatial search and location-based queries ## Core Concepts 1. **Indices and Documents**: Data organized in indices containing JSON documents with unique IDs 2. **Inverted Index**: Data structure enabling fast full-text search by mapping terms to documents 3. **Mapping**: Schema definition defining how documents and fields are stored and indexed 4. **Analyzers**: Tokenization pipelines processing text into searchable terms (standard, custom, language-specific) 5. **Query DSL**: JSON-based query language for complex searches (match, term, range, bool, function_score) 6. **Aggregations**: Framework for computing metrics, buckets, and statistical analysis on data 7. **Sharding and Replication**: Horizontal scaling through index partitioning and data redundancy 8. **集群管理**: Master nodes for coordination, data nodes for storage, ingest nodes for preprocessing 9. **索引生命周期**: Hot-Warm-Cold-Delete architecture for managing time-series data 10. **跨集群搜索**: Cross-cluster search for federated queries across multiple clusters ## Code Examples ### Basic Index Operations and Document Management ```python from elasticsearch import Elasticsearch from elasticsearch.exceptions import NotFoundError, ConflictError from datetime import datetime es = Elasticsearch(["http://localhost:9200"]) def create_product_index(): mapping = { "settings": { "number_of_shards": 3, "number_of_replicas": 1, "analysis": { "analyzer": { "product_analyzer": { "type": "custom", "tokenizer": "standard", "filter": ["lowercase", "asciifolding", "product_synonyms"] } }, "filter": { "product_synonyms": { "type": "synonym", "synonyms": [ "laptop, notebook, portable computer", "phone, smartphone, mobile" ] } } } }, "mappings": { "properties": { "name": {"type": "text", "analyzer": "product_analyzer"}, "description": {"type": "text", "analyzer": "standard"}, "SKU": {"type": "keyword"}, "price": {"type": "float"}, "category": {"type": "keyword"}, "in_stock": {"type": "boolean"}, "tags": {"type": "keyword"}, "created_at": {"type": "date"}, "rating": {"type": "float"}, "location": {"type": "geo_point"} } } } return es.indices.create(index="products", body=mapping) def index_product(product_data): product_doc = { **product_data, "created_at": datetime.utcnow().isoformat() } return es.index(index="products", id=product_data["SKU"], document=product_doc) def get_product(SKU): try: return es.get(index="products", id=SKU) except NotFoundError: return None def update_product(SKU, updates): return es.update(index="products", id=SKU, doc=updates) def delete_product(SKU): return es.delete(index="products", id=SKU) def bulk_index_products(products): operations = [] for product in products: operations.append({"index": {"_index": "products", "_id": product["SKU"]}}) operations.append({**product, "created_at": datetime.utcnow().isoformat()}) return es.bulk(operations=operations, refresh=True) ``` ### Search Queries and Filtering ```python def search_products(query, filters=None, page=1, per_page=20): search_query = { "query": { "bool": { "must": [ {"multi_match": { "query": query, "fields": ["name^3", "description", "tags^2"], "type": "best_fields", "fuzziness": "AUTO" }} ], "filter": [] } }, "from": (page - 1) * per_page, "size": per_page, "sort": [ {"_score": "desc"}, {"rating": "desc"}, {"price": "asc" if filters.get("sort") == "price_asc" else "desc"} ], "highlight": { "fields": { "name": {}, "description": {"fragment_size": 150} } } } if filters: if "category" in filters: search_query["query"]["bool"]["filter"].append({"term": {"category": filters["category"]}}) if "min_price" in filters: search_query["query"]["bool"]["filter"].append({"range": {"price": {"gte": filters["min_price"]}}}) if "max_price" in filters: search_query["query"]["bool"]["filter"].append({"range": {"price": {"lte": filters["max_price"]}}}) if "in_stock" in filters: search_query["query"]["bool"]["filter"].append({"term": {"in_stock": filters["in_stock"]}}) if "tags" in filters: search_query["query"]["bool"]["filter"].append({"terms": {"tags": filters["tags"]}}) return es.search(index="products", body=search_query) def autocomplete(prefix, field="name", limit=10): return es.search(index="products", body={ "query": { "match_phrase_prefix": { field: {"query": prefix, "max_expansions": 50} } }, "size": limit, "_source": ["name", "SKU", "price"] }) def search_with_function_score(query, boost_factors=None): return es.search(index="products", body={ "query": { "function_score": { "query": {"match_all": {}}, "functions": [ {"field_value_factor": {"field": "rating", "factor": 1.2, "modifier": "log1p", "missing": 1}}, {"filter": {"term": {"in_stock": True}}, "weight": 2} ], "score_mode": "sum", "boost_mode": "multiply" } }, "size": 50 }) def fuzzy_search(query, fuzziness="AUTO"): return es.search(index="products", body={ "query": { "multi_match": { "query": query, "fields": ["name", "description"], "fuzziness": fuzziness } } }) def find_similar_products(SKU, limit=10): return es.search(index="products", body={ "query": { "more_like_this": { "fields": ["name", "description", "tags"], "like": [{"_index": "products", "_id": SKU}], "min_term_freq": 1, "min_doc_freq": 1, "max_query_terms": 25 } }, "size": limit }) ``` ### Aggregations and Analytics ```python def get_product_aggregations(category_filter=None): query = {"match_all": {}} if not category_filter else {"term": {"category": category_filter}} return es.search(index="products", body={ "query": query, "size": 0, "aggs": { "categories": {"terms": {"field": "category", "size": 50}}, "price_stats": {"stats": {"field": "price"}}, "price_ranges": { "range": { "field": "price", "ranges": [ {"key": "budget", "to": 50}, {"key": "mid-range", "from": 50, "to": 200}, {"key": "premium", "from": 200} ] } }, "avg_rating_by_category": { "terms": {"field": "category", "size": 20}, "aggs": { "avg_rating": {"avg": {"field": "rating"}} } }, "top_tags": {"terms": {"field": "tags", "size": 30}}, "stock_status": {"terms": {"field": "in_stock"}} } }) def get_sales_analytics(start_date, end_date): return es.search(index="sales", body={ "query": { "range": {"created_at": {"gte": start_date, "lte": end_date}} }, "size": 0, "aggs": { "daily_sales": { "date_histogram": { "field": "created_at", "calendar_interval": "day" }, "aggs": { "total_revenue": {"sum": {"field": "amount"}}, "avg_order_value": {"avg": {"field": "amount"}}, "unique_customers": {"cardinality": {"field": "customer_id"}} } }, "by_category": { "terms": {"field": "category"}, "aggs": { "revenue": {"sum": {"field": "amount"}} } } } }) def get_top_selling_products(limit=10): return es.search(index="sales", body={ "query": {"match_all": {}}, "size": 0, "aggs": { "products": { "terms": {"field": "product_id", "size": limit}, "aggs": { "total_revenue": {"sum": {"field": "amount"}}, "total_quantity": {"sum": {"field": "quantity"}} } } } })