先爆個(gè)坑，我是在WINDWOS上操作的Logstash胰丁，官網(wǎng)上的啟動(dòng)命令是：

cd logstash-5.4.1
bin/logstash -e 'input { stdin { } } output { stdout {} }'

上面是linux的啟動(dòng)命令普筹。windows應(yīng)該這樣啟動(dòng)：

cd logstash-5.4.1
bin/logstash.bat -e 'input { stdin { } } output { stdout {} }'

另一種格式：

以上都是控制臺(tái)手寫再輸入到控制臺(tái)，如何處理大量數(shù)據(jù)呢隘马？

數(shù)據(jù)源shakespeare.json是《莎士比亞》太防。
wjb.conf是配置文件：

input {
  file {
    path => "F:/ElasticSearch/logstash-5.4.1/shakespeare.json/"
    start_position => "beginning"
    ignore_older => 0
    codec => "json"
#    sincedb_path => "/dev/null"
    type => "wjb_log"
  }
}

filter{
    if [bboy_id] {
        mutate { add_tag => "BBOY" }
    }
}

output {
    if [bboy_id] {
        elasticsearch {
            hosts => "localhost:9200"
            index => "bboy-%{+YYYY.MM.dd}"
            document_type => "bboy_log"
        }
    } else {
        elasticsearch {
            hosts => "localhost:9200"
            index => "wjb-%{+YYYY.MM.dd}"
            document_type => "wjb_log"
        }
    }
}

運(yùn)行命令：

wjb@wjb MINGW64 /f/ElasticSearch/logstash-5.4.1
$ bin/logstash.bat -f wjb.conf

通過kibana就可以查看數(shù)據(jù)了。如下：

下面是轉(zhuǎn)載的：

#整個(gè)配置文件分為三部分：input,filter,output
#參考這里的介紹 https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html
input {
  #file可以多次使用酸员，也可以只寫一個(gè)file而設(shè)置它的path屬性配置多個(gè)文件實(shí)現(xiàn)多文件監(jiān)控
  file {
    #type是給結(jié)果增加了一個(gè)屬性叫type值為"<xxx>"的條目蜒车。這里的type，對(duì)應(yīng)了ES中index中的type幔嗦，即如果輸入ES時(shí)酿愧，沒有指定type，那么這里的type將作為ES中index的type邀泉。
    type => "apache-access" 
    path => "/apphome/ptc/Windchill_10.0/Apache/logs/access_log*"
    #start_position可以設(shè)置為beginning或者end嬉挡，beginning表示從頭開始讀取文件钝鸽，end表示讀取最新的，這個(gè)也要和ignore_older一起使用庞钢。
    start_position => beginning
    #sincedb_path表示文件讀取進(jìn)度的記錄拔恰，每行表示一個(gè)文件，每行有兩個(gè)數(shù)字基括，第一個(gè)表示文件的inode颜懊，第二個(gè)表示文件讀取到的位置（byteoffset）。默認(rèn)為$HOME/.sincedb*
    sincedb_path => "/opt/logstash-2.3.1/sincedb_path/access_progress"
    #ignore_older表示了針對(duì)多久的文件進(jìn)行監(jiān)控风皿，默認(rèn)一天河爹，單位為秒，可以自己定制桐款，比如默認(rèn)只讀取一天內(nèi)被修改的文件咸这。
    ignore_older => 604800
    #add_field增加屬性。這里使用了${HOSTNAME}魔眨，即本機(jī)的環(huán)境變量炊苫，如果要使用本機(jī)的環(huán)境變量，那么需要在啟動(dòng)命令上加--alow-env冰沙。
    add_field => {"log_hostname"=>"${HOSTNAME}"}
    #這個(gè)值默認(rèn)是\n 換行符侨艾，如果設(shè)置為空""，那么后果是每個(gè)字符代表一個(gè)event
    delimiter => ""
    #這個(gè)表示關(guān)閉超過（默認(rèn)）3600秒后追蹤文件拓挥。這個(gè)對(duì)于multiline來(lái)說(shuō)特別有用唠梨。... 這個(gè)參數(shù)和logstash對(duì)文件的讀取方式有關(guān)，兩種方式read tail侥啤，如果是read
    close_older => 3600
    coodec => multiline {
      pattern => "^\s"
      #這個(gè)negate是否定的意思当叭，意思跟pattern相反，也就是不滿足patter的意思盖灸。
#      negate => ""
      #what有兩個(gè)值可選 previous和next蚁鳖，舉例說(shuō)明，java的異常從第二行以空格開始赁炎，這里就可以pattern匹配空格開始醉箕，what設(shè)置為previous意思是空格開頭這行跟上一行屬于同一event。另一個(gè)例子徙垫，有時(shí)候一條命令太長(zhǎng)讥裤，當(dāng)以\結(jié)尾時(shí)表示這行屬于跟下一行屬于同一event，這時(shí)需要使用negate=>true姻报，what=>'next'己英。
      what => "previous"
      auto_flush_interval => 60
    }
  }
  file { 
    type => "methodserver-log" 
    path => "/apphome/ptc/Windchill_10.0/Windchill/logs/MethodServer-1604221021-32380.log" 
    start_position => beginning 
    sincedb_path => "/opt/logstash-2.3.1/sincedb_path/methodserver_process"
#    ignore_older => 604800
  }
}
filter{
  #執(zhí)行ruby程序，下面例子是將日期轉(zhuǎn)化為字符串賦予daytag
  ruby {
    code => "event['daytag'] = event.timestamp.time.localtime.strftime('%Y-%m-%d')"
  }
  # if [path] =~ "access" {} else if [path] =~ "methodserver" {} else if [path] =~ "servermanager" {} else {} 注意語(yǔ)句結(jié)構(gòu)
  if [path] =~ "MethodServer" { #z這里的=~是匹配正則表達(dá)式
    grok {
      patterns_dir => ["/opt/logstash-2.3.1/patterns"] #自定義正則匹配
#      Tue 4/12/16 14:24:17: TP-Processor2: hirecode---->77LS
      match => { "message" => "%{DAY:log_weekday} %{DATE_US:log_date} %{TIME:log_time}: %{GREEDYDATA:log_data}"}
    }
    #mutage是做轉(zhuǎn)換用的
    mutate { 
      replace => { "type" => "apache" } #替換屬性值
      convert => { #類型轉(zhuǎn)換
        "bytes" => "integer" #例如還有float
        "duration" => "integer"
        "state" => "integer"
      }
    #date主要是用來(lái)處理文件內(nèi)容中的日期的吴旋。內(nèi)容中讀取的是字符串损肛，通過date將它轉(zhuǎn)換為@timestamp厢破。參考https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html#plugins-filters-date-match
#    date {
#      match => [ "logTime" , "dd/MMM/yyyy:HH:mm:ss Z" ]
#    }
  }else if [type] in ['tbg_qas','mbg_pre'] { # if ... else if ... else if ... else結(jié)構(gòu)
  }else {
    drop{} # 將event丟棄
  }
}
output {
  stdout{ codec=>rubydebug} # 直接輸出，調(diào)試用起來(lái)方便
  # 輸出到redis
  redis {
    host => '10.120.20.208'
    data_type => 'list'
    key => '10.99.201.34:access_log_2016-04'
  }
  # 輸出到ES
  elasticsearch {
    hosts =>"192.168.0.15:9200"
    index => "%{sysid}_%{type}"
    document_type => "%{daytag}"
  }
}