logstashはkafkaを入力源とする

インストール

依存:jdk 7以降
Logstashバージョン:2.3.4
手順:1.公式サイトはtar.をダウンロードしますgzパッケージ
リンクアドレス:リンク2.機械に投げつけて解凍する

tar zxvf logstash-2.3.4.tar.gz

3.kafka inputプラグインをインストールし、ついでにoutputもインストールした

bin/logstash-plugin install logstash-output-kafka
bin/logstash-plugin install logstash-input-kafka

コンフィギュレーション

binフォルダにファイルlogstashを追加する.conf

input {
    kafka {
        zk_connect => "xxx:2181"
        topic_id => "xxx"
        reset_beginning => true
        consumer_threads => 5  
        decorate_events => false 
        }
    }

output {
        elasticsearch {
            action => "index"
            hosts => ["xxx"]
            index => "logs-%{+YYYY.MM.dd}"
            document_type => "logs"
        }
}

【声明:転載は出典独立を明記してください:http://wangnan.tech簡単な本:http://www.jianshu.com/u/244399b1d776 CSDN:http://blog.csdn.net/wangnan9279】

inputにおけるパラメータ解釈

zk_connect

kafka接続zkアドレス、この構成項目でkafkaのデータを取得

group_id

消費者グループは、グループIDによって指定することができ、異なるグループ間の消費は互いに影響を受けず、互いに隔離されている.

topic_id

消費話題を指定し、必須項目でもあり、あるtopicを指定し、これは実はあるテーマを購読し、消費することです.

reset_beginning

logstashが起動するとどの位置からデータの読み取りが開始するかは、デフォルトでは終了位置である、つまりlogstashプロセスは前回の読み取り終了時のオフセット量から読み取りを継続し、以前消費することがなければ最初から読み取りを開始する.既存のデータをインポートする場合は、この設定を「true」に変更すると、logstashプロセスは最初から読み取りを開始します.catに似ていますが、最後の行を読むと終了しません.tail-Fになり、対応するデータの傍受を続けます.

decorate_events

メッセージを出力するときに自身の情報を出力します.消費メッセージのサイズ、topicソース、consumerのグループ情報が含まれます.

rebalance_max_retries

新しいconsumer(logstash)が同じグループに追加されるとreblanceされ、その後partionsの消費者側が新しいconsumerに移行し、あるconsumerがpartitionの消費権限を取得するとzookeeperに登録され、partion Owner registryノード情報になりますが、古いconsumerがこのノードを解放していない可能性があります.この値は制御に使用されます.ノードの再試行回数を登録します.

consumer_timeout_ms

指定された時間内にメッセージが到着しないと例外が投げ出され、通常は変更する必要はありません.
以上は相対的に重要なパラメータの使用例であり、より多くのパラメータはオプションで根拠に従うことができる.https://github.com/joekiller/logstash-kafka/blob/master/README.mdinputのデフォルトパラメータを表示します.

開始

bin/logstash -f logstash.conf &

起動後esに指定したtopicのログデータが書き込まれます
また、esのテンプレートをカスタマイズし、logstashを構成するfilterで必要なビジネスフィールドを切り取ることもできます.ここでは詳細に説明しません.