kafka作为elk缓存使用

ELK集群在大规模的日志收集中面临着数据量大，收集不及时，或宕机的风险，可以选择单节点的redis，但是相比redis，kafka集群高可用的特性，更优，下面来配置kafka集群配置elk作为缓存的方法。

kafka集群的安装配置

一. 初始环境准备

1.服务器准备

主机	地址
db01	10.0.0.200
db02	10.0.0.201
db03	10.0.0.202

# cat /etc/redhat-release  #这里我使用的是centos7.6的系统

CentOS Linux release 7.6.1810 (Core)

2. 下载安装包

安装kafka之前，需要每台服务器上配置好zookeeper

#(以下操作，三台服务器同时操作)

#创建目录

mkdir /kafka

cd /kafka

#下载kafka地址

http://kafka.apache.org/downloads

#下载kafka

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.1/kafka_2.11-2.3.1.tgz

#下载zookeeper地址

http://zookeeper.apache.org/releases.html

#下载zookeeper

wget https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz

#jdk准备好

jdk-8u151-linux-x64.tar.gz

3.配置java环境

#解压tar包

tar xf jdk-8u151-linux-x64.tar.gz -C /opt

#创建软链接

ln -s /opt/jdk1.8.0_151/ /opt/jdk

#配置环境变量

#在/etc/profile后三行添加如下：

# tail -3 /etc/profile

export JAVA_HOME=/opt/jdk

export CLASSPATH=.:$JAVA_HOME/lib:$JAVA_HOME/jre/lib

export PATH=$PATH:$JAVA_HOME/bin

#验证

source /etc/profile 

java -version

java version "1.8.0_151"

二. 安装zookeeper服务

1.安装kafka

#解压zookeeper

tar xf zookeeper-3.4.14.tar.gz -C /opt/

#创建软链接

ln -s /opt/zookeeper-3.4.14 /opt/zookeeper

#创建数据目录

mkdir -p /data/zookeeper

#编辑配置文件

cp /opt/zookeeper/conf/zoo_sample.cfg /opt/zookeeper/conf/zoo.cfg

vim /opt/zookeeper/conf/zoo.cfg

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/data/zookeeper

clientPort=2181

server.1=10.0.0.200:2888:3888

server.2=10.0.0.201:2888:3888

server.3=10.0.0.202:2888:3888

#复制到另外两台

scp /opt/zookeeper/conf/zoo.cfg 10.0.0.201:/opt/zookeeper/conf/

scp /opt/zookeeper/conf/zoo.cfg 10.0.0.202:/opt/zookeeper/conf/

#创建myid

echo 1 > /data/zookeeper/myid  #server1

echo 2 > /data/zookeeper/myid  #server2

echo 3 > /data/zookeeper/myid  #server3

3.启动zookeeper

#三台都启动zookeeper

/opt/zookeeper/bin/zkServer.sh start

#查看状态

# /opt/zookeeper/bin/zkServer.sh status

ZooKeeper JMX enabled by default

Using config: /opt/zookeeper/bin/../conf/zoo.cfg

# Mode: follower 从库

# Mode: leader  主库

#配置环境变量，更方便的使用命令

echo "export PATH=/opt/zookeeper/bin/:$PATH" >> /etc/profile

source /etc/profile

#测试

zkCli.sh -server 10.0.0.202:2181 #可以登录任意一个节点

create /test dhc  #创建

get /test  #查看

set /test ctt #改变key的值

三.安装kafka并测试

1.安装kafka

#以下操作3台服务器都需要操作，以server1为事列

#解压

tar xf kafka_2.11-2.3.1.tgz -C /opt/

cd /opt

#创建软链接

ln -s kafka_2.11-2.3.1 kafka

cd kafka

#编辑配置文件

vim config/server.properties  #我只放修改的，其他的不动

broker.id=1   #对应zookeeper的id

listeners=PLAINTEXT://10.0.0.200:9092 #配置自己的ip地址

log.retention.hours=24   #改成24小时

zookeeper.connect=10.0.0.200:2181,10.0.0.201:2181,10.0.0.202:2181 #配置集群

#测试启动

/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties  #只是测试

#最后一行出现一下，证明启动成功

INFO [KafkaServer id=1] started (kafka.server.KafkaServer)

#ctrl +c掉，重新后台运行

/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

#测试

ps all|grep kafka

#查看日志

tail -f /opt/kafka/logs/server.log

#在node1上创建test

# bin/kafka-topics.sh --create --bootstrap-server 10.0.0.200:9092 --replication-factor 1 --partitions 1 --topic test

#在node2上创建test1

# bin/kafka-topics.sh --create --bootstrap-server 10.0.0.201:9092 --replication-factor 1 --partitions 1 --topic test1

#在node3上查看创建了什么，如果可以看到证明成功

# bin/kafka-topics.sh --list --bootstrap-server 10.0.0.202:9092

#查看具体的topic里面有什么内容

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.200:9092 --topic kafka --from-beginning

#具体实验步骤请看官网

http://kafka.apache.org/quickstart

四安装filebeat并配置

#安装filebeat（这里就不详细说明了，详细请看上一篇博客）

rpm -ivh jdk-8u102-linux-x64.rpm

rpm -ivh filebeat-7.5.0-x86_64.rpm

# grep -Ev '^$|#' /etc/filebeat/filebeat.yml  #修改后的配置文件

filebeat.inputs:

- type: log

  enabled: true

  paths:

    - /var/log/nginx/access.log

filebeat.config.modules:

  path: ${path.config}/modules.d/*.yml

  reload.enabled: false

setup.template.settings:

  index.number_of_shards: 1

setup.kibana:

output.kafka:

  hosts: ["10.0.0.200:9092","10.0.0.201:9092","10.0.0.202:9092"]

  topic: kafka

processors:

  - add_host_metadata: ~

  - add_cloud_metadata: ~

  - add_docker_metadata: ~

  - add_kubernetes_metadata: ~

#启动filebeat

systemctl start filebeat

#验证kafka里面是否有新增topic

/opt/kafka/bin/kafka-topics.sh --list --bootstrap-server 10.0.0.201:9092

#查看topic-kafka里面有没有内容

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.200:9092 --topic kafka --from-beginning

五配置logstash

#安装logstash（具体的安装步骤请看上一篇博客）

rpm -ivh logstash-7.5.0.rpm

#配置logstash

[root@db01 elk]# cat /etc/logstash/conf.d/kafka.conf

input { 

kafka{

		bootstrap_servers => ["10.0.0.200:9092,10.0.0.201:9092,10.0.0.202:9092"]

		group_id => "test"

		auto_offset_reset => "earliest"

		consumer_threads => "5"

		decorate_events => "false"

		topics => ["kafka"]

		type => "bbs_log"

		codec => json

	}

}

output {

stdout {}   #只是用来测试输出

}

巴特西

kafka作为elk缓存使用

ELK集群在大规模的日志收集中面临着数据量大，收集不及时，或宕机的风险，可以选择单节点的redis，但是相比redis，kafka集群高可用的特性，更优，下面来配置kafka集群配置elk作为缓存的方法。

kafka集群的安装配置

一. 初始环境准备

1.服务器准备

2. 下载安装包

3.配置java环境

二. 安装zookeeper服务

1.安装kafka

3.启动zookeeper

三.安装kafka并测试

1.安装kafka

四安装filebeat并配置

五配置logstash

到此，kafka集群配合filebeat+logstash 完成！

最新文章

热门文章

巴特西

kafka作为elk缓存使用

ELK集群在大规模的日志收集中面临着数据量大，收集不及时，或宕机的风险，可以选择单节点的redis，但是相比redis，kafka集群高可用的特性，更优，下面来配置kafka集群配置elk作为缓存的方法。

kafka集群的安装配置

一. 初始环境准备

1.服务器准备

2. 下载安装包

3.配置java环境

二. 安装zookeeper服务

1.安装kafka

3.启动zookeeper

三.安装kafka并测试

1.安装kafka

四 安装filebeat并配置

五 配置logstash

到此，kafka集群配合filebeat+logstash 完成！

最新文章

热门文章

四安装filebeat并配置

五配置logstash