Logstash导入SQL Server至Elasticsearch遇搜索建议及重复导入问题

一、数据重复导入异常（1000行生成365000条文档） #

原因 #

1. 全量重复执行：jdbc输入的schedule => "* * * * *"设置为每分钟执行一次，且SQL语句为SELECT TOP (1000)全量查询，无增量过滤条件，导致每次任务都重复导入相同数据。
2. 文档ID配置滞后：若初期未设置document_id，会生成大量无唯一ID的重复文档；即使后续配置了document_id，旧的无ID文档仍会保留，累积成大量重复数据。

修复方案 #

1. 调整任务触发与查询逻辑
   • 若只需一次性导入，直接删除schedule参数；若需增量同步，修改SQL语句添加增量条件（需表中有时间戳或自增字段，比如LastUpdated）：

     SELECT [Id], [Title], [Teacher], [WebSiteCategory]
     FROM [WebScrapingProject_DB].[dbo].[Test]
     WHERE [LastUpdated] > :sql_last_value
     

     同时在jdbc输入中添加增量配置：

     jdbc {
       # ... 其他原有配置 ...
       use_column_value => true
       tracking_column => "LastUpdated"
       tracking_column_type => "timestamp"
       schedule => "0 * * * *" # 按需调整频率，示例为每小时一次
     }
     

2. 清理现有重复数据
   用Elasticsearch的_delete_by_query API删除无ID的冗余文档：

   curl -X POST "https://localhost:9200/yabland/_delete_by_query" -u elastic:AE1Qj2*vb73VIkpp4Nyh --cacert /home/erfan/SearchHerkunft/logstash-8.10.2/config/http_ca.crt -H "Content-Type: application/json" -d '{
     "query": {
       "bool": {
         "must_not": {
           "exists": { "field": "id" }
         }
       }
     }
   }'
   

3. 确保文档ID唯一性
   当前配置中document_id => "%{[@metadata][_id]}"已基于SQL的Id字段设置唯一ID，需确保该映射正常生效，避免生成无ID文档。

二、搜索建议（suggest字段）配置修复 #

核心问题 #

1. 字段引用错误：filter中%{[@metadata][Title]}是错误引用，Title是事件字段，不在@metadata中，应直接用%{Title}。
2. suggest结构不匹配：Elasticsearch的completion类型需要input数组（可选weight），原配置未正确组装该结构，而是生成了独立的suggest_input、suggest_weight字段。
3. 模板匹配失败：template.json的index_patterns: ["yabland-*"]与目标索引yabland不匹配，导致模板未生效。

修复步骤 #

1. 修正Logstash filter配置
   替换原filter部分，正确组装suggest字段：

   filter {
     mutate {
       copy => { "Id" => "[@metadata][_id]" }
       rename => { "Id" => "id" } # 对应template中的id字段
       remove_field => ["@version"]
     }
     mutate {
       convert => { "[@metadata][_id]" => "string" }
     }
     # 用ruby filter灵活构建suggest结构，自动过滤空值
     ruby {
       code => '
         title = event.get("Title")
         teacher = event.get("Teacher")
         input_list = []
         input_list << title if title && !title.empty?
         input_list << teacher if teacher && !teacher.empty?
         event.set("suggest", { "input" => input_list, "weight" => 10 }) if input_list.any?
       '
     }
   }
   

2. 修正template.json的索引匹配
   将index_patterns改为匹配目标索引：

   {
     "index_patterns": ["yabland"],
     "mappings": {
       "properties": {
         "id": { "type": "keyword" },
         "title": { "type": "text" },
         "teacher": { "type": "text" },
         "webSiteCategory": { "type": "text" },
         "suggest": { "type": "completion", "analyzer": "simple" }
       }
     }
   }
   

3. 手动加载模板（可选）
   若Logstash自动加载模板失败，用API手动上传：

   curl -X PUT "https://localhost:9200/_index_template/yabland-template" -u elastic:AE1Qj2*vb73VIkpp4Nyh --cacert /home/erfan/SearchHerkunft/logstash-8.10.2/config/http_ca.crt -H "Content-Type: application/json" -d @/home/erfan/SearchHerkunft/logstash-8.10.2/config/template.json
   

三、验证修复 #

1. 停止Logstash，清理Elasticsearch目标索引：

   curl -X DELETE "https://localhost:9200/yabland" -u elastic:AE1Qj2*vb73VIkpp4Nyh --cacert /home/erfan/SearchHerkunft/logstash-8.10.2/config/http_ca.crt
   

2. 重启Logstash执行导入：

   bin/logstash -f ~/SearchHerkunft/logstash-8.10.2/config/sqlJDBC.conf
   

3. 验证数据：
   • 检查文档数量是否为1000：

     curl -X GET "https://localhost:9200/yabland/_count" -u elastic:AE1Qj2*vb73VIkpp4Nyh --cacert /home/erfan/SearchHerkunft/logstash-8.10.2/config/http_ca.crt
     

   • 验证suggest字段结构：

     curl -X GET "https://localhost:9200/yabland/_search" -u elastic:AE1Qj2*vb73VIkpp4Nyh --cacert /home/erfan/SearchHerkunft/logstash-8.10.2/config/http_ca.crt -H "Content-Type: application/json" -d '{
       "query": { "match_all": {} },
       "_source": ["suggest"]
     }'
     

内容的提问来源于stack exchange，提问作者sicario