如何用Logstash导出Elasticsearch嵌套字段至GCS CSV文件

一、修复嵌套字段提取为空的问题 #

原配置中mutate过滤器使用点分隔的字段路径（如obj1.Name）无法被Logstash正确识别，需改用方括号嵌套语法引用嵌套字段。同时确保Elasticsearch输入的_source正确指定目标字段：

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", "Hierarchy"],
      "query": { "match_all": {} }
    }'
  }
}

filter {
  # 提取嵌套字段并重命名为CSV列名
  mutate {
    add_field => {
      "col1" => "%{[obj1][Name]}"
      "col2" => "%{[obj1][addr]}"
      "col3" => "%{[obj1][obj2][location]}"
    }
    # 清理原始嵌套字段（可选，减少冗余数据）
    remove_field => ["obj1"]
  }
}

二、提取对象数组中的categoryUrl字段 #

针对Hierarchy数组，分两种常见业务需求处理：

需求1：将所有categoryUrl拼接为单个字段（逗号分隔） #

使用ruby过滤器遍历数组并拼接值：

filter {
  # 承接上方mutate过滤器
  ruby {
    code => '
      hierarchy = event.get("Hierarchy")
      if hierarchy.is_a?(Array)
        category_urls = hierarchy.map { |item| item["categoryUrl"] }.compact.join(",")
        event.set("category_urls", category_urls)
      end
    '
  }
  mutate { remove_field => ["Hierarchy"] }
}

需求2：将数组展开为多条记录（每个数组元素对应一行CSV） #

使用split过滤器拆分数组，每条记录保留原始字段+当前元素的categoryUrl：

filter {
  # 承接上方mutate过滤器
  split { field => "Hierarchy" }
  mutate {
    add_field => { "category_url" => "%{[Hierarchy][categoryUrl]}" }
    remove_field => ["Hierarchy"]
  }
}

三、完整GCS输出配置 #

根据选择的需求，更新CSV codec的columns参数，加入新增字段：

output {
  google_cloud_storage {
    codec => csv {
      include_headers => true
      # 需求1用：["col1", "col2","col3", "category_urls"]
      # 需求2用：["col1", "col2","col3", "category_url"]
      columns => ["col1", "col2","col3", "category_url"]
    }
    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
  }
}

关键注意事项 #

• 提前确认Elasticsearch索引中obj1.obj2.location和Hierarchy字段确实存在，避免因字段缺失导致CSV为空。
• 针对海量数据，可适当调大request_timeout_seconds和flush_interval_secs参数，防止超时或内存溢出。
• 测试阶段可添加stdout { codec => rubydebug }输出到控制台，验证字段提取正确性后再切换到GCS输出。

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