如何使用logstash将多行json对象数组解析为单独的elasticsearch文档?

pqwbnv8z  于 2021-06-10  发布在  ElasticSearch
关注(0)|答案(0)|浏览(314)

我有一个用例,在这个用例中,我将每小时接收一个新文件,这个文件被filebeat提取,然后发送到logstash。该文件在一个数组中包含多个json对象,这些对象被拆分为多行。我想提取数组中的每个对象,并将它们作为单独的文档写入elasticsearch索引。
json的结构如下:

[
    {
        "Results": {
            "StartWeekSec": [
                2000,
                198000.0
            ],
            "ReportStartPeriod": "2019-01-15T07:00:00.000Z",
            "Constellation": "GPS",
            "ReferenceTime": [
                2000,
                31984.00
            ],
            "Status": 0
        },
        "ForecastResults": [
            {
                "Point": [
                    29.67,
                    -7.5
                ],
                "BestAccuracy": 10.98,
                "WorstAccuracy": 14.59              
            },
            {
                "Point": [
                    55.37,
                    54.1
                ],
                "BestAccuracy": 20,
                "WorstAccuracy": 13.99      
            }
        ]   
    },
    {
        "Results": {
            "StartWeekSec": [
                2000,
                198000.0
            ],
            "ReportStartPeriod": "2019-01-15T07:00:00.000Z",
            "Constellation": "GPS",
            "ReferenceTime": [
                2000,
                31984.00
            ],
            "Status": 0
        },
        "ForecastResults": [
            {
                "Point": [
                    29.67,
                    -7.5
                ],
                "BestAccuracy": 10.98,
                "WorstAccuracy": 14.59              
            },
            {
                "Point": [
                    55.37,
                    54.1
                ],
                "BestAccuracy": 20,
                "WorstAccuracy": 13.99      
            }
        ]   
    }   
]

据我所知,文件应该一直是这种格式,但是数组 ForecastResults 其中可以有任意数量的对象,这不是一个定义的数量。
因此,在上面的例子中,我希望创建两个elasticsearch文档,这两个文档都由一个对象组成 Results 以及一组物体 ForecastResults .
我对logstash是个新手,所以我甚至不能百分之百确定这是否可能实现,但我相信应该实现。我试过使用logstash多行插件,但我不知道识别每个文档的模式是什么?
所以我的主要问题是:
从数组中提取每个对象需要什么正则表达式?
提取对象后,如何将其转换为索引Map所需的格式?这可以用grok来完成吗?
我尝试了下面的logstash.conf文件,但它错误地说我不能使用带beats输入的多行编解码器,所以我不知道在这里做什么。我知道我可以在beats方面使用regex,但我仍然不知道regex会是什么样子。

input {
  beats {
    port => 5044
    codec => multiline {
      negate => false
      what => "previous"
      pattern => "{*}"
    }
  }
}
output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "events"
    ilm_enabled => "true"
  }
}

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题