导语

对于一个游戏社区产品，在拥有一定的内容沉淀后，搜索功能作为社区获取内容的最有效途径，是每个社区产品都应该考虑实现的。本文主要介绍基于腾讯云ES如何从零搭建整套社区搜索服务。

需求分析

作为内容社区的相关产品，对应的搜索服务一般需要考虑实现的功能有：

• 社区中各类内容项的搜索结果返回。这里搜索结果的排序评分，除了要考虑对于搜索关键词的匹配评分，根据个业务或者内容项的不同特点，仍需要进行一定修正。基于我们业务所在社区主要包含了以下几类搜索对象：
  1. 官方公告,一些专栏文章,社区帖子.这些内容项的搜索结果排序，主要考虑发表时间以及对应帖子的热度情况，比如查看/评论/点赞的互动数据。
  2. 基于一些特殊的搜索对象，这里要考虑玩家对于搜索对象的喜爱程度，额外要考虑支持基于对象的英文名，中文名，以及中文拼音搜索的匹配。
  3. 对于用户名或一些专栏作者的搜索，主要考虑其粉丝数
• 搜索结果高亮：即搜索结果预览页对于搜索结果关键词的高亮展示
• 搜索建议：根据用户的输入，基于内容库给出对应的搜索建议
• 热搜榜单(或者叫猜你想搜)：基于玩家的历史搜索记录，以及运营的手动调整需求。能够体现和引导当前社区用户的关注点。
• 搜索历史记录，主要有客户端做本地缓存，这里不过多关注

系统整体架构

基于上述的需求分析，使用ES搜索引擎能够完全满足相关的搜索需求，基于此在处理整体搜索后台服务上，主要考虑下面几个问题：

各类搜索数据源从业务的mysql DB数据库导入和同步到ES搜索数据库

一般情况下这里主要有实时同步和定时同步两种方案：

• 实时同步：利用消息队列实时消费mysql的变动的binlog，解析mysql业务数据的实时变动，将实时改动直接同步写入到ES库。比如当前我们使用的CDB也直接提供了关于binlog的增量数据订阅服务，并提供了SDK接入处理。
• 采用定时同步的方式, 基于mysql的定时查询拉取方式，将每次间隔时间内变更的内容批量同步到ES数据库，可以使用logstash组件只需要对应字段映射配置，即可方便的实现。

由于我们业务数据库本身对与数据的处理也没有删除记录的操作，只需要用字段标识状态。且考虑到产品对于实时性要求不高，10分钟级的延迟完全可以接受，最终选择了方案2，利用logstash组件以及对应的字段映射配置，即可实现定时的从业务数据库将需要的搜索数据同步到ES对应的索引中。

玩家行为日志的离线处理

由于业务需要根据用户的行为日志判断搜索对象的受喜欢程度，反过来作用于搜索结果的排序上。所以这里需要将搜索行为，比如点击了某个搜索结果对象的详情操作等，通过客户端上报的灯塔事件记录，进行离线聚合处理计算并将结果字段值导入回ES。

整体架构图

实现细节

基于logstash数据同步

1. logstash简介

logstash 是ELK三个开源项目（ Elasticsearch 、Logstash 、Kibana）中服务器端数据处理管道，能够同时从多个数据来源采集数据，转换数据，然后把数据发送到诸如Elasticsearch但不限于的ES的一些“存储库”中

并且logstash的X-pack功能（高级功能特性，腾讯云白金版支持）能够使该处理管道服务在Kibana中直接接入管理和监控。

2. logstash镜像制作和上云部署

• 镜像制作：由于要访问mysql，而logstash的官方镜像中不包含mysql-connector的jar包，需要基于官方镜像，手动将需要使用的mysql-coonector的jar包放到镜像的 /usr/share/logstash/logstash-core/lib/jars/ 目录下即可使用。
• 上云部署：
• 使用logstash从JDBC连接同步数据的过程是一个定时触发且持续不断的过程，同步过程中我们需要记录上一次同步的offset位置，一般可利用数据表的last_update字段即记录上一次同步的时间偏移点，下一次同步过程只需要查询数据表last_update字段大于offset的数据记录再进行同步即可。利用logstash的dbc-input插件配置对应查询的sql,同步时间间隔和jdbc的相关参数，更多配置细节见 jdbc-input插件文档 。
• 上面的特性决定了我们的服务部署运行过程是依赖一个稳定的持久存储的，用于存储每个数据源表的offset偏移量。使用StatefulSet作为有状态服务的负载对象, 利用K8s的StorageClass动态创建对应的持久存储，腾讯云容器提供了基于云硬盘CBS动态创建的方式，具体可以见文档相关配置： https://cloud.tencent.com/document/product/457/44238 。
• 当前腾讯云好像也提供关于logstash云实例的管理和创建服务，可以直接使用。3. logstash数据同步配置实例

如下配置示例将一个表的定时周期每2分钟同步一次到ES索引当中。更多的配置方法以及函数使用可参考 Logstash文档

input {
  jdbc {
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "{{ MysqlURL }}?characterEncoding=utf8&serverTimezone=Asia/Shanghai"
    jdbc_user => "{{ MysqlUser }}"
    jdbc_password => "{{ MysqlPasswd }}"
    # 设置执行周期参考Cron表达式 ，
    schedule => "*/2 * * * *"
    statement => "SELECT `iId`,`sContent` FROM tbTableName where dtUpdated > :sql_last_value order by dtUpdated asc limit 1000"
    use_column_value => true
    # 是否要把字段名全部小写
    lowercase_column_names => false
    tracking_column_type => "timestamp"
    # 把查询到记录哪个字段作为sql_last_value的值
    tracking_column => "dtLastUpdated"
    last_run_metadata_path => "/usr/share/logstash/last_value/tbTable_sql_last_value.yml"
    type => "CustomType"
filter {
   # filter插件，可以利用其中的一些函数，一处某些字段值不输出到es
  if[type] != "CustomType"{
    mutate {
          remove_field => ["dtLastUpdated"]
  mutate {
          remove_field => ["type"]
output {
    elasticsearch {
        # ES的IP地址及端口
        hosts => "{{ EsHost }}"
        user => "{{ EsUser }}"
        password => "{{ EsPassword }}"
        # 索引名称 可自定义
        index => "index-name"
        # 需要关联的数据库中有有一个id字段，对应类型中的id
        document_id => "%{iId}"
        action => "update"
        doc_as_upsert => true  # 插入或更新选项
    stdout {
        # JSON格式到标准输出，作为日志保存
        codec => json_lines
}

搜索结果优化

1. 使用function_score控制相关度评分

function_score 提供了weight(加权),random_score(随机打分),field_value_factor(使用字段的数值参与计算分数),decay_function(衰减函数 gauss, linear, exp等),script_score(自定义脚本)。这里我们主要使用了gauss衰减函数对内容的产生时间dtLastUpdated进行评分衰减，以及field_value_factor函数对内容的评论数，或者阅读数，进行评分加权。具体查询语句可参考如下：

{
    "query": {
        "function_score": {
            "query": {
                "bool": {
                   "must": {
                        "multi_match": {
                            "query": "查询关键词",
                            "type": "best_fields",
                            "fields": [
                                "sContent.keyword^3",
                                "sContent"
                            "minimum_should_match": "30%"
                    "filter": [
                            "term": {
                                "iStatus": "0"
            "functions": [
                    "gauss": {
                        "dtLastUpdated": {
                            "origin": "now/d",
                            "scale": "360d",
                            "offset": "90d",
                            "decay": 0.5
                    "field_value_factor": {
                        "field": "iCommentCount",
                        "modifier": "ln2p",
                        "factor": 0.05,
                        "missing": 0
}

2. 返回高亮词结果，给前端进行搜索关键词的高亮展示

这里主要利用了es的 highlight ，在搜索语句中添加hight设置，在返回的结果中对于和查询关键词相符的位置直接被特殊的颜色标记，无需前端做特殊处理。

"highlight": {
    "pre_tags": "<span style='color: #6B6CDB'>",
    "post_tags": "</span>",
    "fields": {
        "sContent": {}
}

3. 使用自定义分析器进行拼音搜索

一个分析器就是将三个功能封装到一个里面，三个功能：包括了

• char_filter 字符过滤器，字符串按照顺序通过每个字符过滤器，分词前调整字符串，用来去掉某些特殊字符，或者转换
• tokenizer：分词器， 将字符串分为单个词条，遇到空格或标点或者一些stop word将其拆分未词条
• token_filter：分词过滤器，将tokenizer输出的词条进行修改，删除或者增加 如下示意：使用了ik_smart分词插件以及pinyin过滤器，自定义分词器ik_smart_pinyin

# 自定义分词器
"settings": {
    "analysis": {
        "filter": {
            "my_pinyin": {
                "keep_joined_full_pinyin": "true",
                "padding_char": " ",
                "type": "pinyin",
                "first_letter": "prefix"
        "analyzer": {
            "ik_smart_pinyin": {
                "filter": [
                    "my_pinyin",
                    "word_delimiter"
                "type": "custom",
                "tokenizer": "ik_smart"
}

4. 使用scroll API处理分页问题和优化查询效率

Scroll API 简单来说就是一次性给你所有生成的数据生成了一个快照，并设定保存查询窗口缓存的时间，返回一个scroll_id，拉取下一页时通过scroll_id即可以直接高效的获取下一页的内容。

# 查询语句 将查询上下文保持1分钟，使用scroll=1m表示需要将查询上下文保留一分钟
POST index-nba2app-tbPosts/_search?scroll=1m
"query": {
  "_scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDnF1ZXJ5==",
  "took" : 25,
  "hits":{
# 利用scroll 进行下一页拉取
# 将查询上下文再保存一分钟
GET /_search/scroll
    "scroll": "1m", 
    "scroll_id" : "FGluY2x1ZGVfY29udGV4dF91dWlkDnF1ZXJ5=="
}

热搜榜单

1. 写入搜索日志，直接在业务发起搜索请求时，将搜索关键词作为搜索日志插入一个日志索引即可，该日志索引设置按时间淘汰记录。
2. 使用es聚合查询上述日志索引，得到近一段时间每个关键词的查询次数，结合运营配置的关键词权重加减对结果进行热度merge，排序后作为热搜榜单返回。 如下聚合搜索语句示例：

{
    "query": {
        "bool": {
            "must": [{
                "range": {
                    "logTime": {
                        "gte": "2022-12-04 20:20:18",
                        "lte": "2022-12-04 21:20:18"
                "term": {
                    "zone": {
                        "value": 30
    "aggs": {
        "aggs_name": {
            "terms": {
                "field": "searchContent.keyword",
                "size": 20
}

搜索建议词

使用前缀查询从对应es索引中获取搜索建议，如下示意搜索语句，实际应用中可能包含多个索引内容下建议词结果的合并返回。

POST index_name/_search
"suggest": {
    "custom_suggest_name": {
    "prefix": "【查询关键词前缀】",