[文件行读取JDBC转换并输出到文件] #Logstash

file2file4jdbc.conf

input {
    file {
        path => ["/usr/local/logstash/conf/lbd2tbd/wait_transfer_txt.txt"]
        start_position => "beginning"
        sincedb_path => "/dev/null"
    }
}
filter {

    mutate {
        split  => { "message" => "    "}
    }
    mutate {
        add_field => {"lbdId"=>"%{message[0]}" "viewerId"=>"%{message[1]}"}
    }
    mutate {
       gsub => [ "viewerId", "\r", "" ]
       remove_field => [ "message" ]
    }

    jdbc_streaming {
        jdbc_driver_library => "/usr/local/logstash/sqlserver/sqljdbc4-4.0.jar"
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        jdbc_connection_string => "jdbc:sqlserver://xxxxxxx;DatabaseName=xxxxx"
        jdbc_user => "xxx"
        jdbc_password => "xxxxx"
        statement => "SELECT TBDObjectId FROM xxx.dbo.xxx WHERE xxx = :lbdId;"
        parameters => { "lbdId" => "lbdId"}
        target => "tbdId"
    }

    jdbc_streaming {
        jdbc_driver_library => "/usr/local/logstash/sqlserver/sqljdbc4-4.0.jar"
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
        jdbc_connection_string => "jdbc:sqlserver://xxxx:xxx;DatabaseName=xxxx"
        jdbc_user => "xxx"
        jdbc_password => "xxxx"
        statement => "SELECT UnionId FROM xxx.dbo.xxx WHERE Id = :viewerId"
        parameters => { "viewerId"=> "viewerId"}
        target => "unionId"
    }
    mutate {
        replace => { "tbdId" => "%{[tbdId][0][tbdobjectid]}" "unionId" => "%{[unionId][0][unionid]}" }
    }
}
output {
    stdout {
        codec => rubydebug{}        
    }
    file {
        path => "result.txt"
        codec => line { format => "%{lbdId}>%{tbdId}|%{viewerId}>%{unionId}"}
    }
}