logstash日志写入kafka

• 安装kafka 
  • curl -L -O https://mirrors.cnnic.cn/apache/kafka/0.10.2.1/kafka_2.10-0.10.2.1.tgz
  • tar xf kafka_2.10-0.10.2.1.tgz
  • cd kafka_2.10-0.10.2.1
  • cd config/
  • vim zookeeper.properties 
    • 新加

      host.name=192.168.0.248
      advertised.host.name=192.168.0.248

  • cd ../bin
  • 启动zookeeper   nohup ./zookeeper-server-start.sh ../config/zookeeper.properties &
  • 启动kalfka         nohup ./kafka-server-start.sh ../config/server.properties  &
  • 创建topic  ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic log_test --from-beginning
  • 消费topic   bin/kafka-console-consumer.sh --zookeeper 192.168.0.248:2181 --topic log_test --from-beginning
• 安装logstash
  • wget https://download.elastic.co/logstash/logstash/logstash-2.4.0.tar.gz    tar -zxvf logstash-2.4.0.tar.gz
  • 配置
    • mkdir -p etc
    •  vim etc/www.lanmps.com.conf
      

      • input {
        　　file {
                    type => "nginx-access"
                    path => ["/var/log/nginx/access.log"]
                    start_position => "beginning"
                  }
        }
        output{
                    kafka{
                        topic_id => "log_test"
                        # kafka的地址
                        bootstrap_servers => "192.168.0.248:9092"
                        # 一定要注明输出格式
                       codec => "json"
                 }
        }

        
        
      • 启动logstash

      • nohup bin/logstash -f etc/test.conf &