elk + kafka 简单搭建日志分析系统

迈不过友情╰ 2022-05-31 03:11 546阅读 0赞

# elk + kafka 简单搭建日志分析系统 #

文介绍使用ELK（elasticsearch、logstash、kibana） +　kafka来搭建一个日志系统。

想象一下淘宝，它可以对不同的用户实现推荐不同的产品，它主要是通过分析我们使用淘宝产生的日志来进行推荐的。对于比较小的项目，可以把日志储存到DB中。但是并发量高了怎么办？这时候出现了ELK，但是对于双11这种活动时，那么多人同时访问，单靠logstash传输是不现实的，所以引入kafka消息中间件。还可以用Hadoop实现推荐算法(这个有空在研究).

日志处理流程:使用java把日志传入kafka，然后通过kafka将日志发送给logstash，logstash再将日志写入elasticsearch，这样elasticsearch就有了日志数据了，最后，则使用kibana将存放在elasticsearch中的日志数据显示出来，并且可以做实时的数据图表分析等等。

## 一、配置zookeeper集群 ##

kafka集群需要由zookeeper来管理，所以先搭建zookeeper集群，先下载zookeeper，我使用的是zookeeper-3.4.11.tar.gz，虽然说kafka也有自带zookeeper，但是使用那一个有时候会出一些莫名其妙的问题，建议下载zookeeper配置。  
进入zookeeper-3.4.11.tar.gz所在目录，通过 tar -zxvf ./zookeeper-3.4.11.tar.gz分别解压每台机器上面的zookeeper。  
我的kafka和zookeeper是安装在同一台机器上面的，总共用了三台机器来部署kafka集群，IP分别是:192.168.238.131、192.168.238.132、192.168.238.133. 进入zookeeper-3.4.11，cd zookeeper-3.4.11  
cp ./conf/zoo\_simple.cfg ./conf/zoo.cfg  
vi ./conf/zoo.cfg 分别在每台服务器上面添加添加 dataDir=/tmp/data \#储存数据的  
server.1=192.168.238.131:2888:3888  
server.2=192.168.238.132:2888:3888  
server.3=192.168.238.133:2888:3888  
退出vi编辑器  
server.X=A:B:C  
X-代表服务器编号  
A-代表ip  
B和C-代表端口，这个端口用来系统之间通信  
mkdir /tmp/data  
cd ./data  
添加服务器编号vim ./myid 添加数字1，对应上面的server.X  
启动服务器 ： ./bin/zkServer.sh start  
查看服务器状态 ： ./bin/zkServer.sh status  
查看启动过程 ： ./bin/zkServer.sh start-foreground

## 二、配置kafka集群 ##

配置好zookeeper集群后，开始配置kafka集群，同样使用tar -zxvf kafka\_2.11-1.0.0.tgz解压kafka。  
进入kafka，cd kafka\_2.11-1.0.0  
修改配置文件，  
vi ./config/server.properties:  
broker.id=1  
listeners=PLAINTEXT://:9092  
三台服务器分别设置broker.id=1、broker.id=2、broker.id=3  
分别启动kafka: ./bin/zookeeper-server-start.sh ./config/zookeeper.properties  
创建一个主题 ./bin/kafka-topics.sh –create –zookeeper 192.168.238.132:2181 –replication-factor 1 –partitions 3 –topic test\_topic  
查看kafka主题： ./bin/kafka-topics.sh –list –zookeeper 192.168.238.132:2181

## 三、配置elastisearch ##

我的elastisearch没有集群，elastisearch需要依赖jdk，不能以root用户运行elastisearch。  
  
安装jdk  
tar -zxvf jdk-8u152-linux-x64.tar.gz  
vim /etc/profile  
export JAVA\_HOME=/usr/software/java/jdk1.8.0\_152  
export PATH=$JAVA\_HOME/bin:$PATH  
export CLASSPATH=.:$JAVA\_HOME/lib/dt.jar$JAVA\_HOME/lib/tools.jar  
使配置生效 source /etc/profile  
  
解压elastisearch  
tar -zxvf elasticsearch-6.1.1.tar.gz  
有时候其他机器不能访问es，添加配置  
vim ./config/elasticsearch.yml  
network.host: 0.0.0.0  
http.port: 9200  
启动es：./bin/elasticsearch  
测试es是否启动成功：curl ‘http://192.168.238.137:9200/?pretty’  
192.168.238.137是我es所在机器的IP  
虚拟机上会出现错误：（正式环境不会）  
ERROR: \[1\] bootstrap checks failed  
\[1\]: max virtual memory areas vm.max\_map\_count \[65530\] is too low, increase to at least \[262144\]  
解决：sudo sysctl -w vm.max\_map\_count=262144  
从新启动es：./bin/elasticsearch  
!\[\](https://i.imgur.com/V9rkFLt.jpg)

## 四、配置logstash ##

同样解压logstash，tar -zxvf ./logstash-6.1.1.tar.gz  
同样安装jdk  
在./logstash-6.1.1/config下面新建my.config文件  
  
vi ./logstash-6.1.1/config/my.config  
  
input\{  
kafka\{  
bootstrap\_servers => \[“192.168.238.131:9092,192.168.238.132:9092,192.168.238.133:9092”\]  
group\_id => “group\_logstash”  
id => “my\_kafka\_id”  
topics => \[“test\_topic”\]  
auto\_offset\_reset => latest  
\}  
\}  
output\{  
elasticsearch\{  
hosts => \[“192.168.238.137:9200”\]  
index => “mylog”  
\}  
\}  
  
![jxR3wbX.jpg][]  
退出vi  
  
启动logstash: ./bin/logstash -f ./config/my.config  
  
配置完成

## 五、测试 ##

进入kafka，消费一个  
  
./bin/kafka-console-producer.sh –broker-list 192.168.238.132:9092 –topic test\_topic  
随便输入作为测试  
  
可以在kibana中看见  
  
![ascwqQg.png][]  
也可以用Java程序消费

public static void testProducer () {
        Properties props = new Properties();
        // A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
        props.put("bootstrap.servers", "192.168.238.131:9092,192.168.238.132:9092,192.168.238.133:9092");
        // asks = [all，-1，0，1]  
        // acks=all This means the leader will wait for the full set of in-sync replicas to acknowledge the record. 
        // This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. 
        // This is the strongest available guarantee. This is equivalent to the acks=-1 setting.
        props.put("acks", "all");
        // re try  default 0   value = [0，...，2147483647]
        props.put("retries", 0);
        // default 16384    Buffer size
        props.put("batch.size", 16384);
        // default 0   The upper limit of delay
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("test_topic", "test log : " + Integer.toString(i)));
        }
        producer.close();
    }

![WzojJX7.png][]

[jxR3wbX.jpg]: https://i.imgur.com/jxR3wbX.jpg
[ascwqQg.png]: https://i.imgur.com/ascwqQg.png
[WzojJX7.png]: https://i.imgur.com/WzojJX7.png