ElasticSearch 9.x 从入门到精通实战教程

-- 创建视图：标准化原始日志
CREATE VIEW clean_logs AS
  FROM raw-logs-*
  | RENAME @timestamp AS ts, host.name AS host
  | EVAL level = UPPER(log.level)
  | WHERE level != "DEBUG";

-- 像查询索引一样查询视图
FROM clean_logs
| STATS count = COUNT(*) BY host
| SORT count DESC;

PROMQL index=k8s-downsampled start="2026-02-17T08:00:00Z" end="2026-02-17T09:00:00Z" step=30m
  avg_bytes=(avg(rate(network.total_bytes_in[30m])))
| SORT avg_bytes DESC, step;

// semantic_text 现在默认使用 bfloat16 + bbq_disk
// 无需映射更改；新索引自动获得此优化
PUT my-index
{
  "mappings": {
    "properties": {
      "content": { "type": "semantic_text" }
    }
  }
}

原始文档：
  Doc 1: "ElasticSearch 是搜索引擎"
  Doc 2: "搜索引擎负责全文检索"
  Doc 3: "ElasticSearch 支持向量搜索"

          ↓ 分析器处理（分词 + 归一化）

词元列表：
  Doc 1: [elasticsearch, 是, 搜索, 引擎]
  Doc 2: [搜索, 引擎, 负责, 全文, 检索]
  Doc 3: [elasticsearch, 支持, 向量, 搜索]

          ↓ 构建倒排索引

倒排索引：
  elasticsearch → [Doc1, Doc3]
  搜索          → [Doc1, Doc2, Doc3]
  引擎          → [Doc1, Doc2]
  向量          → [Doc3]
  全文          → [Doc2]
  ...

// 关闭不需要排序/聚合的字段的 doc_values
{
  "mappings": {
    "properties": {
      "session_id": {
        "type": "keyword",
        "doc_values": false  // 仅用于过滤，不需要聚合
      }
    }
  }
}

# 创建 docker 网络
docker network create elastic

# 启动 ElasticSearch 9.4
docker run -d \
  --name es \
  --net elastic \
  -p 9200:9200 \
  -e discovery.type=single-node \
  -e xpack.security.enabled=false \
  -e ES_JAVA_OPTS="-Xms2g -Xmx2g" \
  docker.elastic.co/elasticsearch/elasticsearch:9.4.3

# 启动 Kibana
docker run -d \
  --name kibana \
  --net elastic \
  -p 5601:5601 \
  -e ELASTICSEARCH_HOSTS=http://es:9200 \
  docker.elastic.co/kibana/kibana:9.4.3

# 1. 导入 GPG 密钥
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

# 2. 添加仓库
echo "deb https://artifacts.elastic.co/packages/9.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-9.x.list

# 3. 安装
sudo apt-get update && sudo apt-get install elasticsearch

# 4. 配置 JVM 堆内存（/etc/elasticsearch/jvm.options）
# -Xms16g
# -Xmx16g

# 5. 配置 elasticsearch.yml
cat > /etc/elasticsearch/elasticsearch.yml << 'EOF'
cluster.name: my-cluster
node.name: node-1
network.host: 0.0.0.0
discovery.seed_hosts: ["host1", "host2", "host3"]
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
EOF

# 6. 启动
sudo systemctl enable elasticsearch
sudo systemctl start elasticsearch

# 查看集群健康状态
curl -X GET "localhost:9200/_cluster/health?pretty"

# 响应示例
{
  "cluster_name": "my-cluster",
  "status": "green",           // green=健康, yellow=副本缺失, red=主分片缺失
  "timed_out": false,
  "number_of_nodes": 3,
  "number_of_data_nodes": 3,
  "active_primary_shards": 15,
  "active_shards": 30,
  "relocating_shards": 0,
  "initializing_shards": 0,
  "unassigned_shards": 0,
  "delayed_unassigned_shards": 0
}

shard = hash(routing) % number_of_primary_shards

客户端写入请求
      │
      ▼
协调节点接收 ──→ 根据 routing 计算目标分片
      │
      ▼
主分片所在节点
      │
      ├──→ 1. 写入 Indexing Buffer（内存缓冲区）
      ├──→ 2. 写入 Translog（事务日志，持久化到磁盘）
      │
      ▼  （refresh_interval 到期，默认 1 秒）
      │
  Refresh 操作
      │
      ├──→ 3. Indexing Buffer 生成新的 Segment（内存中）
      ├──→ 4. Segment 刷新到磁盘（但尚未 fsync）
      │
      ▼  新文档变为可搜索
      │
      ▼  （flush 操作，默认每 30 分钟或 translog 达到 512MB）
      │
  Flush 操作
      │
      ├──→ 5. Segment fsync 到磁盘（确保持久化）
      ├──→ 6. 清空 Translog
      │
      ▼  （同时，主分片将写入同步到副本分片）
      │
副本分片所在节点
      │
      └──→ 同样执行写入流程

PUT /products
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1,
    "refresh_interval": "1s"
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": { "type": "keyword" }
        }
      },
      "price": { "type": "scaled_float", "scaling_factor": 100 },
      "category": { "type": "keyword" },
      "tags": { "type": "keyword" },
      "create_time": { "type": "date" },
      "description": { "type": "text", "analyzer": "ik_max_word" }
    }
  }
}

POST /_bulk
{ "index": { "_index": "products", "_id": "1" } }
{ "name": "ElasticSearch 实战指南", "price": 89.90, "category": "图书", "tags": ["技术", "搜索"], "create_time": "2026-06-01" }
{ "index": { "_index": "products", "_id": "2" } }
{ "name": "无线蓝牙耳机 Pro", "price": 599.00, "category": "数码", "tags": ["音频", "无线"], "create_time": "2026-06-15" }
{ "index": { "_index": "products", "_id": "3" } }
{ "name": "机械键盘 RGB 背光", "price": 329.00, "category": "数码", "tags": ["外设", "机械"], "create_time": "2026-07-01" }

GET /products/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "无线 耳机" } }
      ],
      "filter": [
        { "range": { "price": { "lte": 1000 } } }
      ]
    }
  },
  "sort": [
    { "_score": "desc" },
    { "create_time": "desc" }
  ],
  "size": 10
}

┌─────────────────┬──────────┬────────────────────────────────────────┐
│ 类型             │ 大小     │ 适用场景                                │
├─────────────────┼──────────┼────────────────────────────────────────┤
│ byte            │ 1 字节   │ -128 到 127                            │
│ short           │ 2 字节   │ -32768 到 32767                        │
│ integer         │ 4 字节   │ 通用整数，计数器、年龄                  │
│ long            │ 8 字节   │ 大整数，时间戳（内部存储）              │
│ float           │ 4 字节   │ 单精度浮点，对精度要求不高              │
│ double          │ 8 字节   │ 双精度浮点                             │
│ scaled_float    │ 8 字节   │ 金额、评分（long + scaling_factor）     │
│ unsigned_long   │ 8 字节   │ 0 到 2^64-1，无符号长整型               │
└─────────────────┴──────────┴────────────────────────────────────────┘

// 默认行为：每个字符串字段都被双重索引
PUT /logs-default
{ "mappings": { "dynamic": true } }

POST /logs-default/_doc
{
  "service": "api-gateway",      // 被映射为 text + keyword
  "env": "production",           // 被映射为 text + keyword
  "request_id": "abc123",        // 被映射为 text + keyword
  "user_agent": "Mozilla/5.0..." // 被映射为 text + keyword
}
// 结果：4 个字符串字段 × 2 种索引 = 8 份索引数据
// 大部分字段只需要 keyword 精确匹配，text 索引是浪费

// 方案 1：严格模式，拒绝未知字段
PUT /logs-strict
{
  "mappings": {
    "dynamic": "strict"  // 写入未定义字段会报错
  }
}

// 方案 2：动态模板，精确控制类型推断
PUT /logs-template
{
  "mappings": {
    "dynamic": "true",
    "dynamic_templates": [
      {
        "strings_as_keyword": {
          "match_mapping_type": "string",
          "mapping": { "type": "keyword", "ignore_above": 256 }
        }
      }
    ]
  }
}

Character Filter（字符过滤器）
  │  处理原始文本，如去除 HTML 标签
  ▼
Tokenizer（分词器）
  │  将文本拆分为词元
  ▼
Token Filter（词元过滤器）
  │  对词元进行归一化处理
  │  如小写化、词干提取、停用词过滤
  ▼
最终词元列表

输入文本: "ElasticSearch 搜索引擎实战"

ik_max_word: [elasticsearch, 搜索, 搜索引擎, 引擎, 实战]
ik_smart:    [elasticsearch, 搜索引擎, 实战]

PUT /articles
{
  "settings": {
    "analysis": {
      "analyzer": {
        "ik_index": {
          "type": "custom",
          "tokenizer": "ik_max_word",
          "filter": ["lowercase", "asciifolding"]
        },
        "ik_search": {
          "type": "custom",
          "tokenizer": "ik_smart",
          "filter": ["lowercase", "asciifolding"]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "title": {
        "type": "text",
        "analyzer": "ik_index",
        "search_analyzer": "ik_search"
      },
      "content": {
        "type": "text",
        "analyzer": "ik_index",
        "search_analyzer": "ik_search"
      }
    }
  }
}

// 查看分析器输出
POST /articles/_analyze
{
  "analyzer": "ik_max_word",
  "text": "ElasticSearch 搜索引擎实战"
}

// 响应
{
  "tokens": [
    { "token": "elasticsearch", "start_offset": 0, "end_offset": 13, "type": "ENGLISH", "position": 0 },
    { "token": "搜索", "start_offset": 14, "end_offset": 16, "type": "CN_WORD", "position": 1 },
    { "token": "搜索引擎", "start_offset": 14, "end_offset": 18, "type": "CN_WORD", "position": 2 },
    { "token": "引擎", "start_offset": 16, "end_offset": 18, "type": "CN_WORD", "position": 3 },
    { "token": "实战", "start_offset": 18, "end_offset": 20, "type": "CN_WORD", "position": 4 }
  ]
}

PUT /products-synonym
{
  "settings": {
    "analysis": {
      "filter": {
        "my_synonym": {
          "type": "synonym",
          "synonyms": [
            "手机,智能手机,移动电话",
            "耳机,耳麦,耳塞",
            "电脑,计算机,PC"
          ]
        }
      },
      "analyzer": {
        "ik_with_synonym": {
          "type": "custom",
          "tokenizer": "ik_max_word",
          "filter": ["lowercase", "my_synonym"]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "ik_with_synonym"
      }
    }
  }
}

PUT /logs-app
{
  "mappings": {
    "dynamic": "true",
    "dynamic_templates": [
      {
        "strings_as_keyword": {
          "match_mapping_type": "string",
          "mapping": { "type": "keyword", "ignore_above": 256 }
        }
      },
      {
        "longs_as_long": {
          "match_mapping_type": "long",
          "mapping": { "type": "long" }
        }
      },
      {
        "doubles_as_double": {
          "match_mapping_type": "double",
          "mapping": { "type": "double" }
        }
      },
      {
        "dates_as_date": {
          "match": ".*_time$",
          "mapping": { "type": "date" }
        }
      },
      {
        "booleans_as_boolean": {
          "match_mapping_type": "boolean",
          "mapping": { "type": "boolean" }
        }
      }
    ]
  }
}

// 写入的文档
{
  "user": [
    { "first": "张", "last": "三" },
    { "first": "李", "last": "四" }
  ]
}

// ElasticSearch 内部存储（扁平化）
{
  "user.first":  ["张", "李"],
  "user.last":   ["三", "四"]
}

// 查询 first=张 AND last=四 会错误地匹配到这条文档
// 因为 "张" 在 user.first 数组中，"四" 在 user.last 数组中

PUT /users
{
  "mappings": {
    "properties": {
      "user": {
        "type": "nested",
        "properties": {
          "first": { "type": "keyword" },
          "last": { "type": "keyword" }
        }
      }
    }
  }
}

// nested 查询
GET /users/_search
{
  "query": {
    "nested": {
      "path": "user",
      "query": {
        "bool": {
          "must": [
            { "term": { "user.first": "张" } },
            { "term": { "user.last": "三" } }
          ]
        }
      }
    }
  }
}

PUT /company
{
  "mappings": {
    "properties": {
      "relation": {
        "type": "join",
        "relations": {
          "department": "employee"
        }
      }
    }
  }
}

// 写入父文档
PUT /company/_doc/1
{
  "name": "技术部",
  "relation": "department"
}

// 写入子文档（指定父 ID）
PUT /company/_doc/2?routing=1
{
  "name": "张三",
  "relation": { "name": "employee", "parent": "1" }
}

// 查询某部门的所有员工
GET /company/_search
{
  "query": {
    "has_parent": {
      "parent_type": "department",
      "query": { "term": { "name": "技术部" } }
    }
  }
}

// 创建别名
POST /_aliases
{
  "actions": [
    { "add": { "index": "products_v1", "alias": "products" } }
  ]
}

// 通过别名读写（对应用透明）
POST /products/_doc    // 实际写入 products_v1
GET /products/_search  // 实际查询 products_v1

// 零停机重建索引
POST /_aliases
{
  "actions": [
    { "add": { "index": "products_v2", "alias": "products" } },
    { "remove": { "index": "products_v1", "alias": "products" } }
  ]
}

// 写入别名：只指向一个索引
POST /_aliases
{
  "actions": [
    { "add": { "index": "products_v2", "alias": "products_write" } }
  ]
}

// 查询别名：可以指向多个索引
POST /_aliases
{
  "actions": [
    { "add": { "indices": ["products_v1", "products_v2"], "alias": "products_read" } }
  ]
}

// 过滤别名：只暴露部分数据
POST /_aliases
{
  "actions": [
    {
      "add": {
        "index": "products",
        "alias": "products_onsale",
        "filter": { "term": { "status": "on_sale" } }
      }
    }
  ]
}

分片大小        查询延迟（100万文档，match查询）
──────────────────────────────────────────────
  5 GB              8 ms     ← 过多分片，扇出开销大
 10 GB             12 ms
 20 GB             18 ms
 30 GB             25 ms     ← 最优区间
 50 GB             42 ms
 80 GB             85 ms
100 GB            130 ms
150 GB            280 ms     ← 过大，单分片查询慢

// 节点配置（elasticsearch.yml）
// hot 节点：node.attr.data_type: hot
// warm 节点：node.attr.data_type: warm
// cold 节点：node.attr.data_type: cold

// 索引级分片分配过滤
PUT /logs-2026.07/_settings
{
  "index.routing.allocation.require.data_type": "hot"
}

// 通过 ILM 自动迁移
// hot 阶段 → warm 阶段时自动修改 allocation 规则

// 调整磁盘水位线
PUT _cluster/settings
{
  "transient": {
    "cluster.routing.allocation.disk.watermark.low": "85%",
    "cluster.routing.allocation.disk.watermark.high": "90%",
    "cluster.routing.allocation.disk.watermark.flood_stage": "95%"
  }
}

// 批量导入前：关闭 refresh
PUT /products/_settings
{
  "refresh_interval": "-1",
  "number_of_replicas": 0
}

// 执行批量导入... 使用 _bulk API，每批 5-15 MB

// 导入完成后：恢复设置
PUT /products/_settings
{
  "refresh_interval": "1s",
  "number_of_replicas": 1
}

// 手动触发 force merge，减少 Segment 数量
POST /products/_forcemerge?max_num_segments=1

# Python 示例：自适应批量大小
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(["http://localhost:9200"])

def generate_docs():
    for i in range(1000000):
        yield {
            "_index": "products",
            "_id": str(i),
            "_source": {
                "name": f"商品 {i}",
                "price": round(random.uniform(10, 999), 2),
                "category": random.choice(["数码", "图书", "家居"])
            }
        }

# chunk_size 控制每批文档数，max_chunk_bytes 控制每批字节上限
helpers.bulk(es, generate_docs(), chunk_size=1000, max_chunk_bytes=15*1024*1024)

BulkProcessor bulkProcessor = BulkProcessor.builder(
        client,
        new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                // 批量请求前的回调
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                // 批量请求成功后的回调
                if (response.hasFailures()) {
                    logger.warn("Bulk has failures: {}", response.buildFailureMessage());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                // 批量请求失败后的回调
                logger.error("Bulk failed", failure);
            }
        })
    .setBulkActions(1000)           // 每 1000 条触发一次
    .setBulkSize(new ByteSizeValue(15, ByteSizeUnit.MB))  // 每 15MB 触发一次
    .setFlushInterval(TimeValue.timeValueSeconds(5))       // 每 5 秒触发一次
    .setConcurrentRequests(4)       // 允许 4 个并发请求
    .setBackoffPolicy(BackoffPolicy.exponentialBackoff(
        TimeValue.timeValueMillis(100), 3))  // 指数退避重试
    .build();

// 使用
for (Product product : products) {
    bulkProcessor.add(new IndexRequest("products")
        .id(product.getId())
        .source(JSON.toJSONString(product), XContentType.JSON));
}

// 关闭时刷新剩余请求
bulkProcessor.awaitClose(10, TimeUnit.MINUTES);

// 追求写入吞吐：async 模式
PUT /products/_settings
{
  "index.translog.durability": "async",
  "index.translog.sync_interval": "5s",
  "index.refresh_interval": "30s"
}

// 追求数据安全：request 模式（默认）
PUT /products/_settings
{
  "index.translog.durability": "request"
}

// 1. 创建 ILM 策略
PUT _ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_age": "1d",
            "max_primary_shard_size": "50gb"
          },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },
          "set_priority": { "priority": 50 },
          "allocate": {
            "include": { "data_type": "warm" }
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "set_priority": { "priority": 25 },
          "allocate": {
            "include": { "data_type": "cold" }
          }
        }
      },
      "frozen": {
        "min_age": "60d",
        "actions": {
          "searchable_snapshot": { "snapshot_repository": "my-repo" }
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

// 2. 创建组件模板（settings）
PUT _component_template/logs-settings
{
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.lifecycle.name": "logs-policy",
      "index.refresh_interval": "1s",
      "index.routing.allocation.include.data_type": "hot"
    }
  }
}

// 3. 创建组件模板（mappings）
PUT _component_template/logs-mappings
{
  "template": {
    "mappings": {
      "dynamic": "true",
      "dynamic_templates": [
        {
          "strings_as_keyword": {
            "match_mapping_type": "string",
            "mapping": { "type": "keyword", "ignore_above": 512 }
          }
        }
      ],
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "service": { "type": "keyword" },
        "message": { "type": "text" }
      }
    }
  }
}

// 4. 创建索引模板
PUT _index_template/logs-template
{
  "index_patterns": ["logs-*"],
  "data_stream": {},
  "composed_of": ["logs-settings", "logs-mappings"],
  "priority": 200
}

// 5. 写入数据自动创建数据流
POST logs-app/_doc
{ "@timestamp": "2026-07-04T10:00:00Z", "level": "info", "message": "服务启动" }

# 查看数据流
GET _data_stream/logs-app

# 手动滚动数据流
POST logs-app/_rollover

# 修改数据流的 ILM 策略
PUT _data_stream/logs-app
{
  "index_patterns": ["logs-app-*"],
  "data_stream": {},
  "ilm_policy": "new-logs-policy"
}

# 删除数据流（同时删除后备索引）
DELETE _data_stream/logs-app

// 1. 创建新索引
PUT /products_v2
{
  "mappings": {
    "properties": {
      "name": { "type": "text", "analyzer": "ik_max_word" },
      "price": { "type": "scaled_float", "scaling_factor": 100 },
      "category": { "type": "keyword" }
    }
  }
}

// 2. reindex（支持管道转换）
POST /_reindex?wait_for_completion=false
{
  "source": { "index": "products" },
  "dest": { "index": "products_v2" },
  "script": {
    "source": """
      // 可以在 reindex 时做字段转换
      if (ctx._source.price != null) {
        ctx._source.price = ctx._source.price * 1.0;
      }
      // 删除不需要的字段
      ctx._source.remove('internal_id');
    """
  }
}

// 3. 切换别名
POST /_aliases
{
  "actions": [
    { "remove": { "index": "products", "alias": "products" } },
    { "add": { "index": "products_v2", "alias": "products" } }
  ]
}

// 4. 删除旧索引
DELETE /products

// 提高批量大小
POST /_reindex?slices=auto&refresh=false
{
  "source": { "index": "products", "size": 5000 },
  "dest": { "index": "products_v2", "op_type": "create" }
}

// 按时间范围分批 reindex（避免一次性处理过多）
POST /_reindex
{
  "source": {
    "index": "products",
    "query": {
      "range": { "create_time": { "gte": "2026-01-01", "lt": "2026-04-01" } }
    }
  },
  "dest": { "index": "products_v2" }
}

// 好的做法：filter 中的条件稳定且可缓存
GET /products/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "耳机" } }
      ],
      "filter": [
        { "term": { "category": "数码" } },
        { "range": { "price": { "gte": 100, "lte": 2000 } } }
      ]
    }
  }
}

// 避免的做法：将不需要打分的条件放在 must 中
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "耳机" } },
        { "term": { "category": "数码" } },      // 浪费：不需要打分
        { "range": { "price": { "gte": 100 } } }  // 浪费：不被缓存
      ]
    }
  }
}

// 基本匹配：查询文本被分词后做 OR 匹配
GET /products/_search
{
  "query": {
    "match": { "name": "无线蓝牙耳机" }
  }
}

// 要求所有词都匹配（AND）
{
  "query": {
    "match": {
      "name": {
        "query": "无线 蓝牙 耳机",
        "operator": "and"
      }
    }
  }
}

// minimum_should_match 控制最少匹配词数
{
  "query": {
    "match": {
      "description": {
        "query": "高清 无线 蓝牙 降噪 耳机",
        "minimum_should_match": "75%"
      }
    }
  }
}

// 使用 fuzziness 支持拼写纠错
{
  "query": {
    "match": {
      "name": {
        "query": "蓝芽耳机",
        "fuzziness": "AUTO"
      }
    }
  }
}

// "蓝牙耳机" 必须连续出现
{
  "query": {
    "match_phrase": { "name": "蓝牙耳机" }
  }
}

// 允许中间间隔 1 个词
{
  "query": {
    "match_phrase": {
      "name": { "query": "蓝牙 降噪", "slop": 1 }
    }
  }
}

GET /products/_search
{
  "query": {
    "multi_match": {
      "query": "蓝牙耳机",
      "fields": ["name^3", "description^1"],
      "type": "best_fields"
    }
  }
}

GET /products/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "耳机" } }
      ],
      "should": [
        { "match": { "description": "降噪" } },
        { "term": { "tags": "蓝牙" } }
      ],
      "filter": [
        { "range": { "price": { "gte": 100, "lte": 2000 } } },
        { "terms": { "category": ["数码", "音频"] } }
      ],
      "must_not": [
        { "term": { "status": "discontinued" } }
      ],
      "minimum_should_match": 1
    }
  }
}

{
  "query": {
    "bool": {
      "should": [
        {
          "constant_score": {
            "filter": { "term": { "tags": "热销" } },
            "boost": 1.5
          }
        },
        {
          "match": { "name": "耳机" }
        }
      ]
    }
  }
}

// term：精确匹配单个值（不分词）
{ "query": { "term": { "category": "数码" } } }

// terms：匹配多个值（类似 SQL IN）
{ "query": { "terms": { "category": ["数码", "音频", "外设"] } } }

// terms lookup：从另一个文档的字段获取匹配值
{
  "query": {
    "terms": {
      "tags": {
        "index": "user-preferences",
        "id": "user_123",
        "path": "interested_tags"
      }
    }
  }
}

// 数值范围
{ "query": { "range": { "price": { "gte": 100, "lte": 2000 } } } }

// 日期范围（支持数学表达式）
{
  "query": {
    "range": {
      "create_time": {
        "gte": "now-7d/d",    // 7 天前的 0 点
        "lt": "now/d"          // 今天的 0 点
      }
    }
  }
}

// 日期数学表达式示例
"now"           // 当前时间
"now-1h"        // 1 小时前
"now/d"         // 今天的 0 点（舍入到天）
"now+1d/d"      // 明天的 0 点
"2026-07-04||/d"  // 2026-07-04 的 0 点

// 好：舍入到天，可缓存
{ "range": { "create_time": { "gte": "now/d", "lt": "now+1d/d" } } }

// 差：精确到毫秒，每次查询不同，不缓存
{ "range": { "create_time": { "gte": "2026-07-04T10:30:00.123Z" } } }

GET /products/_search
{
  "size": 0,
  "query": {
    "range": { "create_time": { "gte": "2026-06-01" } }
  },
  "aggs": {
    "by_category": {
      "terms": {
        "field": "category",
        "size": 10,
        "order": { "avg_price": "desc" }
      },
      "aggs": {
        "avg_price": { "avg": { "field": "price" } },
        "price_stats": { "stats": { "field": "price" } },
        "price_histogram": {
          "histogram": {
            "field": "price",
            "interval": 200
          }
        }
      }
    }
  }
}

{
  "aggs": {
    "daily_sales": {
      "date_histogram": {
        "field": "create_time",
        "calendar_interval": "1d",
        "format": "yyyy-MM-dd",
        "time_zone": "+08:00",
        "min_doc_count": 0
      },
      "aggs": {
        "revenue": { "sum": { "field": "price" } }
      }
    }
  }
}

{
  "aggs": {
    "monthly": {
      "date_histogram": {
        "field": "create_time",
        "calendar_interval": "1M"
      },
      "aggs": {
        "monthly_revenue": { "sum": { "field": "price" } },
        "cumulative_revenue": {
          "cumulative_sum": { "buckets_path": "monthly_revenue" }
        },
        "mom_growth": {
          "derivative": { "buckets_path": "monthly_revenue" }
        }
      }
    }
  }
}

// composite 聚合：分页获取高基数聚合结果
{
  "size": 0,
  "aggs": {
    "by_user": {
      "composite": {
        "size": 1000,
        "sources": [
          { "user": { "terms": { "field": "user_id" } } },
          { "date": { "date_histogram": { "field": "@timestamp", "calendar_interval": "1d" } } }
        ]
      },
      "aggs": {
        "total_spent": { "sum": { "field": "amount" } }
      }
    }
  }
}

// 响应中包含 after_key，用于获取下一页
// {
//   "aggregations": {
//     "by_user": {
//       "after_key": { "user": "user_999", "date": "2026-07-01" },
//       "buckets": [...]
//     }
//   }
// }

// 下一页查询
{
  "aggs": {
    "by_user": {
      "composite": {
        "size": 1000,
        "sources": [...],
        "after": { "user": "user_999", "date": "2026-07-01" }
      }
    }
  }
}

// 1. 创建 PIT
POST /products/_pit?keep_alive=5m
// 返回: { "id": "pit_id_string" }

// 2. 首次查询
GET /_search
{
  "size": 100,
  "pit": {
    "id": "上一步返回的 pit_id",
    "keep_alive": "5m"
  },
  "query": { "match_all": {} },
  "sort": [
    { "create_time": "asc" },
    { "_shard_doc": "asc" }
  ]
}

// 3. 后续查询：使用上一次结果的 sort 值
GET /_search
{
  "size": 100,
  "pit": {
    "id": "pit_id",
    "keep_alive": "5m"
  },
  "query": { "match_all": {} },
  "sort": [
    { "create_time": "asc" },
    { "_shard_doc": "asc" }
  ],
  "search_after": ["2026-07-01T00:00:00Z", 12345]
}

// 4. 用完删除 PIT
DELETE /_pit
{ "id": "pit_id" }

GET /products/_search?profile=true
{
  "query": {
    "bool": {
      "must": [{ "match": { "name": "耳机" } }],
      "filter": [{ "range": { "price": { "lte": 500 } } }]
    }
  },
  "size": 5
}

{
  "profile": {
    "shards": [
      {
        "id": "[index_name][0][node_id]",
        "searches": [
          {
            "query": {
              "type": "BooleanQuery",
              "description": "+name:耳机 #price:[-∞ TO 500]",
              "time_in_nanos": 145000,
              "breakdown": {
                "create_weight": 5000,      // 创建查询权重
                "build_scorer": 80000,       // 构建 Scorer
                "next_doc": 30000,           // 遍历文档
                "score": 15000,              // 计算得分
                "advance": 5000,             // 跳转文档
                "match_count": 0,
                "compute_max_score": 0,
                "shallow_advance": 10000
              },
              "children": [
                {
                  "type": "TermQuery",
                  "description": "name:耳机",
                  "time_in_nanos": 80000,
                  "breakdown": { ... }
                },
                {
                  "type": "PointRangeQuery",
                  "description": "price:[-∞ TO 500]",
                  "time_in_nanos": 40000,
                  "breakdown": { ... }
                }
              ]
            }
          }
        ]
      }
    ]
  }
}

// 索引映射
PUT /products_suggest
{
  "mappings": {
    "properties": {
      "name": {
        "type": "completion",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "preserve_separators": true,
        "max_input_length": 50
      }
    }
  }
}

// 写入建议数据
POST /products_suggest/_doc
{
  "name": {
    "input": ["无线蓝牙耳机", "蓝牙耳机", "无线耳机"],
    "weight": 10
  }
}

// 查询建议
GET /products_suggest/_search
{
  "suggest": {
    "product-suggest": {
      "prefix": "蓝牙",
      "completion": {
        "field": "name",
        "size": 5,
        "skip_duplicates": true
      }
    }
  }
}

// 通过 REST API 执行 ES|QL
POST /_query?format=txt
{
  "query": """
    FROM products
    | WHERE price > 100 AND price < 1000
    | KEEP name, price, category
    | SORT price DESC
    | LIMIT 20
  """
}

// 全文匹配
FROM products
| WHERE name : "蓝牙耳机"
| KEEP name, price, category
| SORT price ASC
| LIMIT 10

// 使用 match 函数（支持更多参数）
FROM products
| WHERE match(name, "降噪 耳机")
| EVAL score = score(name)
| SORT score DESC
| LIMIT 10

// 多字段匹配
FROM products
| WHERE name : "耳机" OR description : "降噪"
| KEEP name, price
| LIMIT 20

// 基础聚合
FROM products
| STATS
    count = COUNT(*),
    avg_price = AVG(price),
    min_price = MIN(price),
    max_price = MAX(price),
    total_value = SUM(price)
  BY category
| SORT avg_price DESC

// 多层聚合
FROM products
| WHERE create_time >= "2026-06-01"
| EVAL month = DATE_FORMAT("yyyy-MM", create_time)
| STATS
    revenue = SUM(price),
    items = COUNT(*)
  BY month, category
| SORT month ASC, revenue DESC

// 使用 sparkline 生成迷你图表（9.4 新增）
FROM products
| STATS price_trend = sparkline(price) BY category
| KEEP category, price_trend

// 使用 LIMIT BY 限制每组行数（9.4 Tech Preview）
FROM products
| SORT price DESC
| LIMIT 3 BY category

-- 创建视图：标准化原始日志
CREATE VIEW clean_logs AS
  FROM raw-logs-*
  | RENAME @timestamp AS ts, host.name AS host
  | EVAL level = UPPER(log.level)
  | WHERE level != "DEBUG"
  | KEEP ts, host, level, message;

-- 像查询索引一样查询视图
FROM clean_logs
| STATS count = COUNT(*) BY host
| SORT count DESC
| LIMIT 10

-- 视图可以与物理索引混合查询
FROM clean_logs, alerts-*
| WHERE level == "ERROR"
| KEEP ts, host, message, alert_type
| SORT ts DESC
| LIMIT 50

// 查看数据流中的指标列表
TS k8s-metrics
| METRICS_INFO
| WHERE metric_type == "counter"
| SORT metric_name

// 查看每个时间序列的维度
TS k8s-metrics
| TS_INFO
| SORT metric_name, dimensions

// 计算速率（counter rate）
TS k8s-metrics
| STATS avg_rate = AVG(RATE(network.bytes, 5m)) BY TBUCKET(10m), host
| SORT host, TBUCKET(10m)

// 9.4 新增：窗口大小不必是 bucket 的整数倍
TS k8s-metrics
| STATS AVG(RATE(requests, 15m)) BY TBUCKET(10m), host

// GROK 解析非结构化日志
FROM nginx-logs
| GROK message "%{IP:client_ip} - %{DATA:user} \[%{TIMESTAMP_ISO8601:ts}\] \"%{WORD:method} %{URIPATH:path} %{DATA:protocol}\" %{NUMBER:status} %{NUMBER:bytes}"
| KEEP client_ip, method, path, status, bytes
| WHERE status >= 400
| SORT ts DESC
| LIMIT 20

// DISSECT 解析结构化日志（比 GROK 快）
FROM app-logs
| DISSECT message "%{service} [%{level}] %{timestamp} %{msg}"
| EVAL level = UPPER(level)
| WHERE level == "ERROR"

// JSON 提取
FROM events
| EVAL user_agent = JSON_EXTRACT(headers, "user-agent")
| EVAL ip = JSON_EXTRACT(request, "remote_addr")

// 9.4 新增函数
FROM web-logs
| USER_AGENT user_agent           // 解析 UA 为结构化字段
| REGISTERED_DOMAIN host          // 提取注册域名
| URI_PART url                    // 提取 URI 各部分

写入路径：
  原始文档 → 嵌入模型(text-embedding) → 向量(dense_vector) → 量化(bbq_disk) → HNSW 索引

查询路径：
  查询文本 → 嵌入模型(相同模型) → 查询向量 → kNN 搜索(HNSW + 量化) → Top-K 结果

// 1. 创建推理端点（使用内置模型或外部 API）
PUT _inference/text_embedding/my-embedding
{
  "service": "openai",
  "service_settings": {
    "model_id": "text-embedding-3-small",
    "api_key": "sk-..."
  }
}

// 2. 创建索引，使用 semantic_text 字段
PUT /knowledge-base
{
  "mappings": {
    "properties": {
      "content": {
        "type": "semantic_text",
        "inference_id": "my-embedding"
      },
      "title": { "type": "text" },
      "category": { "type": "keyword" }
    }
  }
}

// 3. 写入文档（自动生成向量）
POST /knowledge-base/_doc
{
  "title": "ElasticSearch 性能调优",
  "content": "通过调整分片数量和映射配置可以显著提升查询性能...",
  "category": "技术"
}

// 4. 语义搜索
GET /knowledge-base/_search
{
  "query": {
    "semantic": {
      "field": "content",
      "query": "如何优化搜索速度"
    }
  }
}

// 部署 ELSER 模型
PUT _ml/trained_models/.elser_model_2
{
  "description": "ELSER v2 model"
}

// 启动模型部署
POST _ml/trained_models/.elser_model_2/deployment/_start?wait_for=starting

// 创建推理端点
PUT _inference/text_embedding/elser-endpoint
{
  "service": "elser",
  "service_settings": {
    "num_allocations": 1,
    "num_threads": 4
  }
}

// 使用 ELSER 的 semantic_text
PUT /knowledge-base-elser
{
  "mappings": {
    "properties": {
      "content": {
        "type": "semantic_text",
        "inference_id": "elser-endpoint"
      }
    }
  }
}

PUT /articles-vector
{
  "mappings": {
    "properties": {
      "title": { "type": "text" },
      "embedding": {
        "type": "dense_vector",
        "dims": 1536,
        "index": true,
        "similarity": "cosine",
        "index_options": {
          "type": "bbq_disk",
          "m": 16,
          "ef_construction": 100
        }
      }
    }
  }
}

// kNN 查询
GET /articles-vector/_search
{
  "knn": {
    "field": "embedding",
    "query_vector": [0.1, 0.2, /* 1536 维向量 */],
    "k": 10,
    "num_candidates": 100
  },
  "_source": ["title"]
}

GET /knowledge-base/_search
{
  "query": {
    "bool": {
      "should": [
        {
          "semantic": {
            "field": "content",
            "query": "如何优化搜索速度"
          }
        },
        {
          "match": {
            "title": "性能 调优 优化"
          }
        }
      ]
    }
  },
  "size": 10
}

GET /knowledge-base/_search
{
  "size": 10,
  "retriever": {
    "rrf": {
      "retrievers": [
        {
          "standard": {
            "query": {
              "semantic": {
                "field": "content",
                "query": "如何优化搜索速度"
              }
            }
          }
        },
        {
          "standard": {
            "query": {
              "match": {
                "title": "性能 调优 优化"
              }
            }
          }
        }
      ],
      "rank_window_size": 50,
      "rank_constant": 60
    }
  }
}

// 先用 kNN 取 50 个候选，再用 rerank 精排取 top 10
GET /knowledge-base/_search
{
  "size": 10,
  "knn": {
    "field": "content.embedding",
    "query_vector_builder": {
      "text_embedding": {
        "model_id": "my-embedding",
        "model_text": "如何优化搜索速度"
      }
    },
    "k": 50,
    "num_candidates": 500
  },
  "rescore": {
    "window_size": 50,
    "query": {
      "rescore_query": {
        "text_expansion": {
          "content": {
            "model_id": ".elser_model_2",
            "model_text": "如何优化搜索速度"
          }
        }
      }
    }
  }
}

// 配置量化位深度（9.4 新增）
{
  "properties": {
    "embedding": {
      "type": "dense_vector",
      "dims": 1536,
      "index_options": {
        "type": "bbq_disk",
        "config": {
          "bits": 4    // 可选 1, 2, 4, 7
        }
      }
    }
  }
}

// 带过滤的 kNN 查询
{
  "knn": {
    "field": "embedding",
    "query_vector": [/* ... */],
    "k": 10,
    "num_candidates": 100,
    "filter": {
      "bool": {
        "filter": [
          { "term": { "category": "技术" } },
          { "range": { "publish_date": { "gte": "2026-01-01" } } }
        ]
      }
    }
  }
}

# elasticsearch.yml 配置
node.attr.ml.gpu: true

# jvm.options 配置
-Xms16g
-Xmx16g
-XX:+UseG1GC
-XX:MaxGCPauseMillis=200
-XX:G1ReservePercent=25

# 或通过环境变量
ES_JAVA_OPTS="-Xms16g -Xmx16g -XX:+UseG1GC"

// 好的做法：filter 中的条件稳定且可缓存
GET /products/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "耳机" } }
      ],
      "filter": [
        { "term": { "category": "数码" } },
        { "range": { "price": { "gte": 100, "lte": 2000 } } }
      ]
    }
  }
}

// 避免的做法：将不需要打分的条件放在 must 中
{
  "query": {
    "bool": {
      "must": [
        { "match": { "name": "耳机" } },
        { "term": { "category": "数码" } },      // 浪费：不需要打分
        { "range": { "price": { "gte": 100 } } }  // 浪费：不被缓存
      ]
    }
  }
}

// 避免：script_score 全表扫描
{
  "query": {
    "script_score": {
      "query": { "match_all": {} },
      "script": { "source": "doc['popularity'].value * doc['recency'].value" }
    }
  }
}

// 替代方案：在写入时计算好排序字段
// 文档写入时增加 rank_score 字段
{ "name": "耳机", "rank_score": 8.5, "popularity": 100, "recency": 0.085 }

// 查询时直接用 function_score 或 sort
{
  "query": { "match": { "name": "耳机" } },
  "sort": [{ "rank_score": "desc" }]
}

// 使用 script 的正确姿势
{
  "query": {
    "bool": {
      "filter": [
        { "range": { "price": { "gte": 100 } } }  // 先用 filter 缩小范围
      ]
    }
  },
  "script_fields": {
    "custom_score": {
      "script": {
        "source": "doc['popularity'].value * params.weight",
        "params": { "weight": 1.5 }  // 使用 params 避免编译
      }
    }
  }
}

// 好：舍入到天，可缓存
{ "range": { "create_time": { "gte": "now/d", "lt": "now+1d/d" } } }

// 差：精确到毫秒，每次查询不同，不缓存
{ "range": { "create_time": { "gte": "2026-07-04T10:30:00.123Z" } } }

PUT /logs-app
{
  "settings": {
    "index.sort.field": ["@timestamp"],
    "index.sort.order": ["desc"]
  }
}

场景：10 亿条日志，查询最近 10 条
────────────────────────────────────
无预排序：     ~3000 ms（扫描所有匹配文档后排序取 top 10）
有预排序：       ~5 ms（按已排序顺序读取 10 条即终止）
提升：        600 倍

# elasticsearch.yml 节点角色配置

# 专用主节点（3 个，保证选举多数）
node.roles: [master]

# 数据节点（根据数据温度细分）
node.roles: [data, data_content, data_hot, data_warm, data_cold]

# 协调节点（不存数据，只路由和合并）
node.roles: []

# ingest 节点（数据预处理）
node.roles: [ingest]

// 查看线程池状态
GET /_cat/thread_pool?v&h=node_name,name,active,queue,rejected,completed

// 关键线程池
// search: 查询线程池，queue 满会拒绝查询请求
// write: 写入线程池，queue 满会拒绝写入请求
// merge: Segment 合并线程池

// 调整搜索线程池大小
PUT _cluster/settings
{
  "transient": {
    "thread_pool.search.queue_size": 2000,
    "thread_pool.search.size": 20
  }
}

// 调整熔断器
PUT _cluster/settings
{
  "transient": {
    "indices.breaker.total.limit": "70%",
    "indices.breaker.fielddata.limit": "40%",
    "indices.breaker.request.limit": "60%"
  }
}

// 开启慢查询日志
PUT /products/_settings
{
  "index.search.slowlog.threshold.query.warn": "2s",
  "index.search.slowlog.threshold.query.info": "500ms",
  "index.search.slowlog.threshold.fetch.warn": "1s",
  "index.search.slowlog.threshold.fetch.info": "200ms",
  "index.indexing.slowlog.threshold.index.warn": "1s",
  "index.indexing.slowlog.threshold.index.info": "200ms"
}

[2026-07-04T10:30:15,123][WARN][index.search.slowlog.query] [node-1] [products][0] took[1.5s], took_millis[1500], total_hits[1000000], types[], stats[], search_type[QUERY_THEN_FETCH], total_shards[3], source[{"query":{"bool":{"must":[{"match":{"name":{"query":"耳机"}}}]}}}],

# 获取热点线程
GET /_nodes/hot_threads?threads=10&interval=500ms&snapshots=10

# 输出示例
# node-1
#   85.3% [cpu] com.elasticsearch.search.SearchService.executeQuery()
#   10.2% [cpu] org.apache.lucene.search.BooleanQuery$BooleanWeight.scorer()
#    4.5% [cpu] com.lucene.index.SegmentReader.document()

查询慢
  │
  ├── 是否深度分页？
  │     ├── 是 → 改用 search_after / PIT
  │     └── 否 ↓
  │
  ├── filter 是否在 must 中？
  │     ├── 是 → 移到 filter 子句
  │     └── 否 ↓
  │
  ├── 是否有脚本查询？
  │     ├── 是 → 预计算字段替代脚本
  │     └── 否 ↓
  │
  ├── 聚合是否高基数？
  │     ├── 是 → 用 composite 或 cardinality
  │     └── 否 ↓
  │
  ├── Segment 数量是否过多？
  │     ├── 是 → force merge 只读索引
  │     └── 否 ↓
  │
  ├── 是否使用 profile 定位？
  │     └── 使用 _search?profile=true 找到最慢子查询
  │
  └── 检查集群负载
        ├── GC 停顿是否频繁 → 调整 JVM
        ├── 磁盘 I/O 是否饱和 → 增加 hot 节点
        └── 线程池 rejected 是否高 → 扩容或优化

写入慢
  │
  ├── refresh_interval 是否过短？
  │     └── 批量写入时设为 -1 或 30s
  │
  ├── Bulk 批次是否太小？
  │     └── 调整到 5-15 MB
  │
  ├── 副本同步是否成为瓶颈？
  │     └── 批量导入时临时设为 0
  │
  ├── Segment 合并是否落后？
  │     └── 检查 _cat/segments，调整合并策略
  │
  ├── translog 是否过大？
  │     └── 检查 index.translog.flush_threshold_size
  │
  └── 检查磁盘 I/O
        └── SSD vs HDD，IOPS 是否充足

PUT /ecommerce
{
  "settings": {
    "number_of_shards": 5,
    "number_of_replicas": 1,
    "refresh_interval": "1s",
    "analysis": {
      "analyzer": {
        "pinyin_analyzer": {
          "tokenizer": "pinyin_tokenizer"
        }
      },
      "tokenizer": {
        "pinyin_tokenizer": {
          "type": "pinyin",
          "keep_first_letter": true,
          "keep_full_pinyin": true,
          "lowercase": true
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "product_id": { "type": "keyword" },
      "name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "fields": {
          "keyword": { "type": "keyword", "ignore_above": 256 },
          "pinyin": { "type": "text", "analyzer": "pinyin_analyzer" },
          "completion": { "type": "completion" }
        }
      },
      "price": { "type": "scaled_float", "scaling_factor": 100 },
      "category": { "type": "keyword" },
      "brand": { "type": "keyword" },
      "tags": { "type": "keyword" },
      "sales_count": { "type": "integer" },
      "rating": { "type": "float" },
      "status": { "type": "keyword" },
      "create_time": { "type": "date" },
      "description": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart"
      }
    }
  }
}

GET /ecommerce/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "蓝牙耳机",
            "fields": ["name^5", "name.pinyin^3", "description^1"],
            "type": "best_fields",
            "minimum_should_match": "70%"
          }
        }
      ],
      "should": [
        { "term": { "tags": "热销" } },
        { "range": { "rating": { "gte": 4.5 } } }
      ],
      "filter": [
        { "term": { "status": "on_sale" } },
        { "range": { "price": { "gte": 50, "lte": 3000 } } }
      ],
      "minimum_should_match": 0
    }
  },
  "functions": [
    {
      "filter": { "range": { "sales_count": { "gte": 1000 } } },
      "weight": 1.5
    }
  ],
  "sort": [
    { "_score": "desc" },
    { "sales_count": "desc" },
    { "create_time": "desc" }
  ],
  "aggs": {
    "categories": { "terms": { "field": "category", "size": 10 } },
    "brands": { "terms": { "field": "brand", "size": 10 } },
    "price_ranges": {
      "range": {
        "field": "price",
        "ranges": [
          { "to": 100 },
          { "from": 100, "to": 500 },
          { "from": 500, "to": 1000 },
          { "from": 1000 }
        ]
      }
    }
  },
  "size": 20
}

// 搜索补全查询
GET /ecommerce/_search
{
  "suggest": {
    "product-suggest": {
      "prefix": "蓝",
      "completion": {
        "field": "name.completion",
        "size": 10,
        "skip_duplicates": true
      }
    }
  }
}

// 同时返回搜索结果和补全建议
GET /ecommerce/_search
{
  "query": { "match_all": {} },
  "suggest": {
    "product-suggest": {
      "prefix": "蓝牙",
      "completion": {
        "field": "name.completion",
        "size": 5
      }
    }
  },
  "size": 0
}

# Python 示例：从数据库同步商品到 ElasticSearch
import psycopg2
from elasticsearch import Elasticsearch, helpers

es = Elasticsearch(["http://localhost:9200"])
pg = psycopg2.connect("dbname=shop user=postgres")

def sync_products():
    cursor = pg.cursor()
    cursor.execute("""
        SELECT id, name, price, category, brand, tags, sales_count,
               rating, status, create_time, description
        FROM products
        WHERE update_time > NOW() - INTERVAL '1 hour'
    """)

    actions = []
    for row in cursor:
        action = {
            "_index": "ecommerce",
            "_id": row[0],
            "_source": {
                "product_id": row[0],
                "name": row[1],
                "price": row[2],
                "category": row[3],
                "brand": row[4],
                "tags": row[5],
                "sales_count": row[6],
                "rating": row[7],
                "status": row[8],
                "create_time": row[9],
                "description": row[10],
                "name_completion": {
                    "input": [row[1], row[1].split()],
                    "weight": row[6]  # 按销量设置权重
                }
            }
        }
        actions.append(action)

    helpers.bulk(es, actions, chunk_size=500)
    print(f"Synced {len(actions)} products")

PUT _index_template/app-logs
{
  "index_patterns": ["app-logs-*"],
  "data_stream": {},
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.lifecycle.name": "logs-policy",
      "index.refresh_interval": "1s",
      "index.mapping.total_fields.limit": 2000,
      "index.sort.field": ["@timestamp"],
      "index.sort.order": ["desc"]
    },
    "mappings": {
      "dynamic": "true",
      "dynamic_templates": [
        {
          "strings_as_keyword": {
            "match_mapping_type": "string",
            "mapping": { "type": "keyword", "ignore_above": 512 }
          }
        }
      ],
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "service": { "type": "keyword" },
        "host": { "type": "keyword" },
        "message": { "type": "text", "analyzer": "standard" },
        "trace_id": { "type": "keyword" },
        "duration_ms": { "type": "integer" }
      }
    }
  }
}

// 按服务统计错误率和 P99 延迟
FROM app-logs
| WHERE @timestamp >= NOW() - 1 hours
| STATS
    total = COUNT(*),
    errors = COUNT(WHERE level == "ERROR"),
    error_rate = errors * 100.0 / total,
    p99_latency = PERCENTILE(duration_ms, 99),
    avg_latency = AVG(duration_ms)
  BY service
| SORT error_rate DESC

// 查找慢请求的调用链
FROM app-logs
| WHERE duration_ms > 1000 AND level != "DEBUG"
| KEEP @timestamp, service, trace_id, message, duration_ms
| SORT duration_ms DESC
| LIMIT 50

// 错误趋势分析
FROM app-logs
| WHERE level == "ERROR" AND @timestamp >= NOW() - 24 hours
| EVAL hour = DATE_TRUNC(1 hour, @timestamp)
| STATS error_count = COUNT(*) BY hour, service
| SORT hour ASC, service

// Top 10 错误消息
FROM app-logs
| WHERE level == "ERROR"
| STATS count = COUNT(*) BY message
| SORT count DESC
| LIMIT 10

// 使用 Kibana Alerting 配置告警条件
GET /app-logs/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        { "range": { "@timestamp": { "gte": "now-5m/m" } } },
        { "terms": { "level": ["ERROR", "FATAL"] } }
      ]
    }
  },
  "aggs": {
    "by_service": {
      "terms": { "field": "service", "size": 20 },
      "aggs": {
        "error_count": { "value_count": { "field": "@timestamp" } },
        "sample_messages": {
          "top_hits": {
            "size": 3,
            "_source": ["message", "trace_id"],
            "sort": [{ "@timestamp": "desc" }]
          }
        }
      }
    }
  }
}

// 创建 Ingest Pipeline
PUT _ingest/pipeline/app-logs-pipeline
{
  "description": "应用日志预处理",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{TIMESTAMP_ISO8601:log_time} \\[%{LOGLEVEL:level}\\] \\[%{DATA:service}\\] \\[%{DATA:trace_id}\\] %{GREEDYDATA:msg}"
        ]
      }
    },
    {
      "date": {
        "field": "log_time",
        "formats": ["ISO8601"],
        "target_field": "@timestamp"
      }
    },
    {
      "convert": {
        "field": "duration_ms",
        "type": "integer",
        "ignore_missing": true
      }
    },
    {
      "script": {
        "source": """
          if (ctx.duration_ms != null && ctx.duration_ms > 2000) {
            ctx.slow_request = true;
          }
        """
      }
    },
    {
      "remove": {
        "field": "log_time",
        "ignore_missing": true
      }
    }
  ],
  "on_failure": [
    {
      "set": {
        "field": "error.message",
        "value": "Pipeline failed: {{ _ingest.on_failure_message }}"
      }
    }
  ]
}

// 在索引设置中指定默认 pipeline
PUT _component_template/app-logs-pipeline
{
  "template": {
    "settings": {
      "index.default_pipeline": "app-logs-pipeline"
    }
  }
}

// 配置远程集群
PUT _cluster/settings
{
  "persistent": {
    "cluster": {
      "remote": {
        "cluster_east": {
          "seeds": ["east-node1:9300", "east-node2:9300"]
        },
        "cluster_west": {
          "seeds": ["west-node1:9300", "west-node2:9300"]
        }
      }
    }
  }
}

// 跨集群搜索
GET /cluster_east:logs-*,cluster_west:logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "range": { "@timestamp": { "gte": "now-1h" } } },
        { "term": { "level": "ERROR" } }
      ]
    }
  },
  "size": 50
}

// 跨集群 ES|QL 查询
POST /_query
{
  "query": """
    FROM cluster_east:app-logs, cluster_west:app-logs
    | WHERE level == "ERROR" AND @timestamp >= NOW() - 1 hours
    | STATS error_count = COUNT(*) BY service
    | SORT error_count DESC
  """
}

# elasticsearch.yml
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.http.ssl.enabled: true

# 生成证书
# elasticsearch-certutil ca
# elasticsearch-certutil cert --ca elastic-stack-ca.p12

# 创建用户和角色

// 创建自定义角色
POST /_security/role/logs_reader
{
  "indices": [
    {
      "names": ["app-logs-*"],
      "privileges": ["read", "view_index_metadata"]
    }
  ]
}

// 创建用户
POST /_security/user/log_viewer
{
  "password": "secure_password",
  "roles": ["logs_reader"],
  "full_name": "Log Viewer"
}

// 创建 API Key（推荐用于服务间认证）
POST /_security/api_key
{
  "name": "log-collector",
  "role_descriptors": {
    "log_writer": {
      "indices": [
        {
          "names": ["app-logs-*"],
          "privileges": ["write", "create_doc", "create_index"]
        }
      ]
    }
  }
}

入门                进阶                高级                优化                专家                生产
 │                  │                  │                  │                  │                  │
 ├─ 概念+CRUD       ├─ 映射+DSL        ├─ 聚合+分页       ├─ 索引+查询调优   ├─ ES|QL+向量搜索  ├─ 集群运维+监控
 ├─ 倒排索引        ├─ 分析器+分词     ├─ 管道聚合        ├─ JVM 调优        ├─ 混合搜索        ├─ 安全配置
 ├─ 分片+副本       ├─ 动态模板        ├─ PIT+search_after├─ Segment 合并    ├─ RAG 重排序      ├─ 跨集群搜索
 └─ REST API        └─ nested/join     └─ composite 聚合  └─ 索引预排序      └─ GPU 加速        └─ ILM 冷热分层