[logstash use http rest] #Logstash #Http

use-http.conf

input {
    jdbc {
        jdbc_driver_library => "/usr/local/logstash/sqlserver/sqljdbc4-4.0.jar"    
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        jdbc_connection_string => "jdbc:sqlserver://xxx:xxx;DatabaseName=xxx"
        jdbc_user => "xxx"
        jdbc_password => "xxx"
        jdbc_paging_enabled => "true"
        jdbc_fetch_size => 200000
        jdbc_page_size => 200000
        statement => " SELECT j.Id as job_id, j.Name AS title, j.Location as location, f.Name AS company, j.EmergencyDegree as is_emcy, j.created as job_created, r.created as rec_created FROM xxx.dbo.Jobs j LEFT JOIN xxx.dbo.Firms f ON j.FirmId = f.Id LEFT JOIN xxx.dbo.Recommendations r ON j.id = r.JobId"
        #parameters => { "id" => 3510 }
		codec => plain{ charset => "GBK" }
    }
}
filter {
    jdbc_streaming {
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "jdbc:mysql://192.168.1.164:3306/xxx?characterEncoding=UTF-8"
        jdbc_user => "xxx"
        jdbc_password => "xxx"
        statement => "SELECT location_name_cn FROM dict_location WHERE location_id = :codeParam"
        parameters => { "codeParam" => "location"}
        target => "location"
    }

    mutate {
        replace => { "location" => "%{[location][0][location_name_cn]]}" "type" => 1}
    }
    rest {
        request => {
            url => "http://192.xxx1.31:9205/xxx/xxx"
            method => "post"
            headers => { "Content-Type" => "application/json" }
            # 注意 此处的body，每一个key对应的value必须通过'%{}'指定，否则rest不会解析
            body => [ {"type"=> "%{type}" "name" => "%{company}"} ]
        }
        json => true
        target => "standard_company"
    }
    rest {
        request => {
            url => "http://192.168.xxx.31:8680/xxxx"
            method => "post"
            headers => { "Content-Type" => "application/json" }
            body => [ {"title"=> "%{title}" "key" => "%{jobid}"} ]
        }
        json => true
        target => "standard_title"
    }
    mutate {
        replace => { 
            "standard_company" => "%{[standard_company][data][0][standard_name]]}"
            "standard_title" => "%{[standard_title][data][0][title_prediction_position]}"
        }
    }

}
output {
    stdout {
        codec => rubydebug{}        
    }
    jdbc {
        driver_jar_path => "/usr/local/logstash/mysql/mysql-connector-java-5.1.47.jar"
        driver_class => "com.mysql.jdbc.Driver"
        connection_string => "jdbc:mysql://192.168.1.164:3306/xxx?useUnicode=true&characterEncoding=utf8"
        username => "xxx"
        password => "xxx"
        max_pool_size => 10
        statement => ["INSERT INTO active_jobs_logstash (job_id,title, standard_title,company,standard_company,location,is_emcy,job_created,rec_created) VALUES (?,?,?,?,?,?,?,?,?)", "job_id", "title", "standard_title", "company", "standard_company", "location", "is_emcy", "job_created", "rec_created"]
    }

}