Logstash

Question

所以目前我正在分析来自 MySQL 字幕数据库的数据，并将它们放入 ElasticSearch 5.2 中。无论如何，我的 ES logstash 具有以下过滤器：

filter {
    grok {
           match => ["subtitles", "%{TIME:[_subtitles][start]} --> %{TIME:[_subtitles][end]}%{GREEDYDATA:[_subtitles][sentence]}" ]
          }
}

产生以下结果：

"_subtitles": {
                  "sentence": [
                     "im drinking latte",
                     "im drinking coffee",
                     "while eating a missisipi cake"
                  ],
                  "start": [
                     "00:00:00.934",
                     "00:00:01.934",
                     "00:00:04.902"
                  ],
                  "end": [
                     "00:00:02.902",
                     "00:00:03.902",
                     "00:00:05.839"
                  ]
               }

但我想要的是：

 "_subtitles": [
                     {
                          "sentence": "im drinking latte",
                          "start": "00:00:00.934",
                          "end": "00:00:02.902"
                       },
                     {... same structure as above},
                     {... same structure as above},
]

请记住，_subtitles 将通过预定义的映射进行嵌套。

原始数据如下：

00:00:00.934 --> 00:00:02.902
im drinking latte

00:00:01.934 --> 00:00:03.902
im drinking coffee

00:00:04.902 --> 00:00:05.839
while eating a missisipi cake

如何使用 Grok 的匹配模式和占位符实现此目的？

Answer 1

经过大量研究和阅读，我找到了答案

我发现最好的方法是： - 离开 Logstash 并编写我自己的脚本以从 mysql 迁移到 Elastic，但随后我必须进行所有模式识别和替换，这可能会变得有些复杂。 - post-使用 Ruby script/filter.

处理字段

解决方法如下：

ruby {
      code => "
        subtitles = []
        starts = event.get('start')
        ends = event.get('end')
        sentences = event.get('sentence')
        counter = 0
        starts.each do |v|
         temp_hash = {}
         temp_hash['index'] = counter
         temp_hash['start'] = v
         temp_hash['end'] = ends[counter]
         temp_hash['sentence'] = sentences[counter]
         counter += 1
         subtitles.push(temp_hash)
        end
        event.set('subtitles', subtitles)
      "
  }

希望对您有所帮助。

但现在我正在尝试改进这一点，因为我的 ElasticSearch 容器因 "cannot handle requests"/ 之类的东西而失败了一段时间......只是因为索引（目前来自 mysql 的大约 20k 行）到 Elastic 中，每个大约有 40 个嵌套对象。

我可以做些什么来加快速度？

也许是一种标记文档的方法，这样我就不会处理它们并将它们标记为前一天或某天已处理？

谢谢，问候。

Answer 2

我想说更好的方法是先使用 split filter，用 split { terminator => "\n\n" } 将内容拆分为单独的短语事件，然后使用 grok（一次拆分单个字幕).

Logstash - 使用 Grok 匹配模式将结果分离到不同的对象中

Logstash - Separate results into different objects using Grok match pattern

elasticsearch

logstash-grok

经过大量研究和阅读，我找到了答案