如何使用 Logstash 将 Elasticsearch 索引中的嵌套字段作为 CSV 文件导出到 Google Cloud Storage

Question

我正在使用 ElasticSearch，在这里我们正在创建按天计算的索引，并且每分钟都会摄取大量数据。想将每天创建的索引中的几个字段导出到 Google 云存储，我能够从索引中获取直接字段，如何使用 Logstash 从弹性搜索索引中的嵌套对象获取字段并将它们作为 csv 文件发送到 GCS 存储桶

尝试在 conf 下从索引中获取嵌套字段，它不起作用并在输出 csv 文件中给出空值：

input {

 elasticsearch {

    hosts => "host:443"

    user => "user"

    ssl => true

    connect_timeout_seconds => 600

    request_timeout_seconds => 600

    password => "pwd"

    ca_file => "ca.crt"

    index => "test"

    query => '

    {
    "_source": ["obj1.Name","obj1.addr","obj1.obj2.location"],

    "query": {

    "match_all": {}

    }

    }

  '

  }

}

filter {
mutate {
    rename => {
        "obj1.Name" => "col1"
        "obj1.addr" => "col2"
        "obj1.obj2.location" => "col3"
    }
  }
 }


output {
   google_cloud_storage {
   codec => csv {
    include_headers => true
    columns => [ "col1", "col2","col3"]
   }
     bucket => "bucket"
     json_key_file => "creds.json"
     temp_directory => "/tmp"
     log_file_prefix => "log_gcs"
     max_file_size_kbytes => 1024
     date_pattern => "%Y-%m-%dT%H:00"
     flush_interval_secs => 600
     gzip => false
     uploader_interval_secs => 600
     include_uuid => true
     include_hostname => true
   }
}

如何从对象数组中获取填充到 csv 上方的字段，在下面的示例中我想获取类别网址:

"Hierarchy" : [
            {
              "level" : "1",
              "category" : "test",
              "categoryUrl" : "testurl1"
            },
            {
              "level" : "2",
              "category" : "test2",
              "categoryUrl" : "testurl2"
            }}

Answer 1

你需要使用Logstash field notation

mutate {
    rename => {
        "[obj1][Name]" => "col1"
        "[obj1][addr]" => "col2"
        "[obj1][obj2][location]" => "col3"
    }
  }
 }