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

Question

因此，目前我正在分析来自我的MySQL字幕db的数据，并将它们放在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迁移到弹性，但然后我必须做所有的模式识别和替换，这可能会变得有些复杂。-使用Ruby脚本/过滤器对字段进行后处理。

解决办法如下：

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容器失败了，比如“无法处理请求”/运行了一段时间。仅仅是因为索引(目前从mysql到Elastic中大约有20k行)，每个嵌套对象大约有40个。

我能做些什么让它更快吗？

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

谢谢，问候。

Answer 2

我认为更好的方法是首先使用分裂滤波器，将内容拆分为与split { terminator => "\n\n" }分离的短语-事件，然后使用grok (这将在一次拆分单个字幕)。