一、环境准备与依赖安装
要实现基于业务权重的商品排名提升，最稳健的技术方案是利用Elasticsearch的Painless脚本进行自定义评分。首先需要搭建一个单节点的ES环境用于本地验证。请确保你的系统已安装Docker，直接执行以下命令启动ES 8.11.0版本：
```bash
docker run -d \
--name es-shop \
-p 9200:9200 \
-p 9300:9300 \
-e "discovery.type=single-node" \
-e "xpack.security.enabled=false" \
-e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
docker.elastic.co/elasticsearch/elasticsearch:8.11.0
```
等待约30秒服务启动后，我们需要准备Python环境来操作ES。创建一个项目目录，并新建requirements.txt文件，填入以下依赖：
```text
elasticsearch==8.11.0
```
在终端执行安装命令：
```bash
pip install -r requirements.txt
```
二、定义商品索引结构
为了让排名算法生效，索引必须包含用于计算权重的数值字段。我们将建立一个包含sales（销量）、stock（库存）、price（价格）和is_new（是否新品）的商品索引。在Kibana的Dev Tools或通过curl执行以下Mapping创建语句：
```json
PUT /products
{
"mappings": {
"properties": {
"product_id": { "type": "keyword" },
"title": {
"type": "text",
"analyzer": "ik_max_word",
"fields": {
"keyword": { "type": "keyword" }
}
},
"category": { "type": "keyword" },
"price": { "type": "double" },
"sales": { "type": "integer" },
"stock": { "type": "integer" },
"is_new": { "type": "boolean" },
"create_time": { "type": "date" }
}
}
}
```
注意：这里使用了ik_max_word分词器，如果你使用的是标准Docker镜像，需要额外安装IK插件。为了简化操作，本指南使用ES自带的standard分词器即可满足演示需求，请将上述JSON中的analyzer字段删除。
三、批量写入模拟数据
我们需要写入一些具有明显特征差异的数据，以便后续验证排序效果。创建一个Python文件init_data.py，写入以下完整代码。这段代码会生成10条商品数据，其中包含高销量低利润、低销量高利润等不同组合：
```python
from elasticsearch import Elasticsearch
from datetime import datetime
es = Elasticsearch("http://localhost:9200")
准备模拟数据
products = [
{"product_id": "P001", "title": "高性能无线鼠标 办公游戏通用", "price": 99.0, "sales": 5000, "stock": 100, "is_new": False},
{"product_id": "P002", "title": "机械键盘 青轴 办公专用", "price": 299.0, "sales": 200, "stock": 50, "is_new": True},
{"product_id": "P003", "title": "4K高清显示器 27英寸 IPS屏", "price": 1999.0, "sales": 50, "stock": 20, "is_new": True},
{"product_id": "P004", "title": "USB-C 扩展坞 多接口转换器", "price": 159.0, "sales": 800, "stock": 200, "is_new": False},
{"product_id": "P005", "title": "人体工学椅 网布透气 可调节", "price": 899.0, "sales": 100, "stock": 10, "is_new": False},
{"product_id": "P006", "title": "无线蓝牙耳机 降噪长续航", "price": 399.0, "sales": 3000, "stock": 500, "is_new": True},
{"product_id": "P007", "title": "智能手表 运动健康监测", "price": 699.0, "sales": 150, "stock": 80, "is_new": False},
{"product_id": "P008", "title": "便携式SSD硬盘 1TB 高速", "price": 450.0, "sales": 20, "stock": 5, "is_new": True},
{"product_id": "P009", "title": "笔记本支架 铝合金 散热", "price": 59.0, "sales": 6000, "stock": 1000, "is_new": False},
{"product_id": "P010", "title": "高清网课摄像头 1080P自动对焦", "price": 129.0, "sales": 10, "stock": 0, "is_new": False},
]
actions = []
for p in products:
actions.append({
"_index": "products",
"_id": p["product_id"],
"_source": {
p,
"create_time": datetime.now()
}
})
批量插入
from elasticsearch.helpers import bulk
success, failed = bulk(es, actions)
print(f"成功插入 {success} 条数据")
```
执行python init_data.py完成数据准备。
四、编写自定义评分脚本
这是实现商品排名提升的核心。我们不仅要匹配文本相关性（BM25），还要结合业务权重。我们的目标是：优先展示销量高、库存充足且是新品的商品。
我们将使用Painless脚本编写一个评分函数。计算逻辑如下：
- 基础分：ES默认的文本相关性得分（_score）。
- 销量权重：销量的对数值，防止销量过大导致分数爆炸。
- 库存惩罚：如果库存为0，分数大幅降低。
- 新品加权：如果是新品，额外加分。
对应的查询DSL结构如下，请重点关注script_score部分：
```json
GET /products/_search
{
"query": {
"function_score": {
"query": {
"match": {
"title": "无线"
}
},
"script_score": {
"script": {
"lang": "painless",
"source": """
double score = _score;
double sales = doc['sales'].value;
double stock = doc['stock'].value;
boolean isNew = doc['is_new'].value;
// 1. 销量权重：取对数并乘以系数，避免数值过大
double salesWeight = Math.log(sales + 1) 2.0;
// 2. 库存逻辑：无库存商品降权（乘以0.1）
double stockFactor = (stock > 0) ? 1.0 : 0.1;
// 3. 新品加权：新品额外加5分
double newBonus = isNew ? 5.0 : 0.0;
// 最终得分计算
return (score + salesWeight + newBonus) stockFactor;
"""
}
},
"boost_mode": "replace"
}
}
}
```
技术细节解析：
boost_mode: replace表示完全使用脚本计算的返回值作为最终排序依据，不再与原始_score相加。这能让你更精确地控制排序逻辑。Math.log(sales + 1)是处理数值型权重的常用技巧，确保数据平滑。
五、Python查询代码实现
为了方便在业务代码中调用，我们将上述DSL封装为Python函数。新建search_products.py：
```python
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200")
def search_with_ranking(keyword):
query = {
"query": {
"function_score": {
"query": {
"match": {
"title": keyword
}
},
"script_score": {
"script": {
"lang": "painless",
"source": """
double score = _score;
double sales = doc['sales'].value;
double stock = doc['stock'].value;
boolean isNew = doc['is_new'].value;
double salesWeight = Math.log(sales + 1) 2.0;
double stockFactor = (stock > 0) ? 1.0 : 0.1;
double newBonus = isNew ? 5.0 : 0.0;
return (score + salesWeight + newBonus) stockFactor;
"""
}
},
"boost_mode": "replace"
}
}
}
resp = es.search(index="products", body=query)
print(f"搜索关键词: {keyword}, 总命中: {resp['hits']['total']['value']}")
print("-" 80)
for hit in resp['hits']['hits']:
source = hit['_source']
print(f"商品: {source['title']} | 销量: {source['sales']} | 价格: {source['price']} | 综合得分: {hit['_score']:.2f}")
if __name__ == "__main__":
测试搜索“无线”，观察P001(高销量)和P006(高销量+新品)的排名
search_with_ranking("无线")
```
六、结果验证与调优
执行python search_products.py。预期输出中，高性能无线鼠标（销量5000）和无线蓝牙耳机（销量3000且是新品）应该排在最前面。尽管高清网课摄像头标题也包含“无线”，但由于其销量只有10且库存为0，它的得分会被stockFactor严重惩罚，排名会非常靠后甚至消失。
如果发现排名未按预期变化，请检查以下几点：
- 字段类型：确保
sales和stock在Mapping中是数值类型（integer/long），如果是text类型会导致脚本报错。
- 空值处理：如果某些商品缺少
sales字段，脚本执行时会抛出异常。更健壮的写法是使用doc['sales'].size() == 0 ? 0 : doc['sales'].value来进行非空判断。
- 性能监控：脚本评分会消耗CPU资源。在生产环境，如果数据量达到千万级，建议将计算好的“分数”预计算好存入文档字段，直接利用该字段排序，而不是实时计算。
通过上述步骤，你已经构建了一个完整的、可落地的商品智能排序系统。这套逻辑不依赖任何第三方付费组件，完全基于开源技术栈实现，能够直接应用于电商搜索、内容推荐等需要精细化控制排名的场景。
