logstash rpm安裝 以及配置文件修改

1.下載相應的版本

wget https://artifacts.elastic.co/downloads/logstash/logstash-6.2.3.rpm

2.執行安裝命令

rpm -ivh logstash-6.2.3.rpm

3.配置文件修改

在/etc/logstash下面 修改內存以及日誌文件地址

在/etc/logstash/conf.d裏面添加本身的配置文件

例子：

input {
    mongodb {
       uri => "mongodb://IP:host/dbname"
        placeholder_db_dir => "/home/mongodb/db/"#同步指針存儲的地方
        placeholder_db_name =>"tablename"
        collection => "message"
        type=>"yt_message"
        batch_size => 10000#每次10000條
    }
}

input {
    mongodb {
        uri => "mongodb://IP:host/dbname"
        placeholder_db_dir => "/home/mongodb/db/"
        placeholder_db_name =>"tablename.db"
        collection => "comment"
        type=>"yt_comment"
        batch_size => 5000
    }
}

filter{
    if[type] =~ "yt_aspect"{
        if "," in [aspect]{
            mutate { 
                rename => ["_id", "uid"]
                remove_field => "log_entry"
                convert => ["bid", "string"]
                split => ["aspect",","]
                add_field => {
                    "aspect1" => "%{[aspect][0]}"
                }
                add_field => {
                    "aspect2" => "%{[aspect][1]}"
                }
            }    
        }else{
            mutate { 
                rename => ["_id", "uid"]
                remove_field => "log_entry"
                convert => ["bid", "string"]
                add_field => {
                    "aspect1" => ""
                }
                add_field => {
                    "aspect2" => ""
                }
            }    
        }       
    }else{
        mutate { 
            rename => ["_id", "uid"]
            remove_field => "log_entry"
            convert => ["bid", "string"]
        }
    }
    ruby{
        code => "
            className=event.get('type')

            if className.include?'message'
                then
                    event.set('flag',{'name' => event.get('flag')})
                end

            if className.include?'comment'
                then
                    event.set('flag',{'name' => event.get('flag'),'parent' => event.get('pid')})
                end
             if className.include?'weibo'
                then
                    event.set('flag',{'name' => event.get('flag'),'parent' => event.get('pid')})
                end

        "
    }
    #date {
    #    match => ["cTime", "yyyy-MM-dd HH:mm:ss Z"]
    #    target => "cTime"
    #}
    
}

output {
    file {
            path => "/home/log/logstash/logstash%{+yyyy.MM.dd}.log"
        }
        stdout {
           codec => json_lines
    }

    if[type] == "yt_message"{
        elasticsearch {
            hosts => ["ip:host"]
            index => "%{projectId}"
            document_type => "%{projectId}"
            document_id => "%{unique}"
            routing => "%{projectId}"
        }
    }

    if[type] == "yt_comment"{         elasticsearch {             hosts => ["IP:host"]             index => "%{projectId}"             document_type => "%{projectId}"             manage_template=>true             document_id => "%{unique}"             routing => "%{projectId}"         }     }   }