如何根据文件名给ElasticSearch设置不同的Logstash索引名

要实现将in-N.csv的数据存入payment_test-N索引，核心是从输入文件的路径中提取数字后缀，再动态构造ES索引名，具体修改步骤如下：

1. 在Filter阶段提取文件后缀数字 #

Logstash的file输入会自动为每条事件添加path字段（包含文件完整路径），我们需要从这个字段中提取文件名里的数字部分，存入自定义字段（比如file_suffix）。

可以用grok插件实现匹配提取，在原filter块中添加以下配置：

filter {
    csv {
        separator => ";"
        columns => ["name","deposit","month"]
    }
    mutate {
        convert => {
            "deposit" => "integer"
        }
    }
    # 新增：从path字段提取文件名中的数字后缀
    grok {
        match => { "path" => ".*in-(?<file_suffix>\d+)\.csv$" }
        # 避免匹配失败时添加额外标签，可选配置
        tag_on_failure => []
    }
}

• 正则.*in-(?<file_suffix>\d+)\.csv$会匹配路径末尾的in-N.csv格式，将数字N捕获到file_suffix字段中。
• 如果需要排查匹配问题，可以去掉tag_on_failure => []，此时匹配失败的事件会被打上_grokparsefailure标签。

2. 动态配置Elasticsearch索引名 #

在output的elasticsearch配置中，将index参数改为动态引用file_suffix字段：

output {
    elasticsearch {
        hosts => "http://localhost:9200"
        index => "payment_test-%{file_suffix}"
    }
    stdout {}
}

这样每条事件都会根据自身的file_suffix值，存入对应的payment_test-N索引。

完整配置示例 #

input {
    file {
        path => "C:/elastic/logstash-8.3.2/config/*.csv"
        start_position => "beginning"
        sincedb_path => "NULL"
    }
}
filter {
    csv {
        separator => ";"
        columns => ["name","deposit","month"]
    }
    mutate {
        convert => {
            "deposit" => "integer"
        }
    }
    grok {
        match => { "path" => ".*in-(?<file_suffix>\d+)\.csv$" }
        tag_on_failure => []
    }
}
output {
    elasticsearch {
        hosts => "http://localhost:9200"
        index => "payment_test-%{file_suffix}"
    }
    stdout {}
}

验证方式 #

启动Logstash后，查看stdout输出的事件，确认file_suffix字段已正确提取；同时可以调用ES的_cat/indices接口，或通过Kibana查看索引列表，检查payment_test-0、payment_test-1等索引是否被创建并写入数据。

内容的提问来源于stack exchange，提问作者Yağız Can Aslan