实验 elk+filebeat+kafka

kafka 3.4.1

elk+filebeat+kafka

实现日志收集

httpd1 mysql1 topic 2.7 3.0

关闭防火墙

systemctl stop firewalld

systemctl disable firewalld

setenforce 0

安装 JDK

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel

java -version

安装 Zookeeper

cd /opt

tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz

mv apache-zookeeper-3.5.7-bin /opt/zookeeper

修改配置文件

cd /opt/zookeeper/conf/

cp zoo_sample.cfg zoo.cfg

tickTime=2000 #通信心跳时间,Zookeeper服务器与客户端心跳时间,单位毫秒

initLimit=10 #Leader和Follower初始连接时能容忍的最多心跳数(tickTime的数量),这里表示为10*2s

syncLimit=5 #Leader和Follower之间同步通信的超时时间,这里表示如果超过52s,Leader认为Follwer死掉,并从服务器列表中删除Follwer

dataDir=/opt/zookeeper/data ●修改,指定保存Zookeeper中的数据的目录,目录需要单独创建

dataLogDir=/opt/zookeeper/logs ●添加,指定存放日志的目录,目录需要单独创建

clientPort=2181 #客户端连接端口

#添加集群信息

server.1=192.168.176.70:3188:3288

server.2=192.168.176.71:3188:3288

server.3=192.168.176.72:3188:3288

在每个节点上创建数据目录和日志目录

mkdir /opt/zookeeper/data

mkdir /opt/zookeeper/logs

在每个节点的dataDir指定的目录下创建一个 myid 的文件

echo 1 > /opt/zookeeper/data/myid

echo 2 > /opt/zookeeper/data/myid

echo 3 > /opt/zookeeper/data/myid

配置 Zookeeper 启动脚本

vim /etc/init.d/zookeeper

#!/bin/bash

#chkconfig:2345 20 90

#description:Zookeeper Service Control Script

ZK_HOME='/opt/zookeeper'

case $1 in

start)

echo "---------- zookeeper 启动 ------------"

$ZK_HOME/bin/zkServer.sh start

;;

stop)

echo "---------- zookeeper 停止 ------------"

$ZK_HOME/bin/zkServer.sh stop

;;

restart)

echo "---------- zookeeper 重启 ------------"

$ZK_HOME/bin/zkServer.sh restart

;;

status)

echo "---------- zookeeper 状态 ------------"

$ZK_HOME/bin/zkServer.sh status

;;

*)

echo "Usage: $0 {start|stop|restart|status}"

esac

设置开机自启

chmod +x /etc/init.d/zookeeper

chkconfig --add zookeeper

分别启动 Zookeeper

service zookeeper start

查看当前状态

service zookeeper status

部署 kafka 集群

安装:

cd /opt/

tar zxvf kafka_2.13-2.7.1.tgz

mv kafka_2.13-2.7.1 kafka

修改配置文件

cd kafka/config/

cp server.properties server.properties.bak

vim server.properties

改经纪人id

如果修改了id 28行可以不改

broker.id=0 ●21行,broker的全局唯一编号,每个broker不能重复,因此要在其他机器上配置 broker.id=1、broker.id=2

listeners=PLAINTEXT://192.168.80.10:9092 ●31行,指定监听的IP和端口,如果修改每个broker的IP需区分开来,也可保持默认配置不用修改

num.network.threads=3 #42行,broker 处理网络请求的线程数量,一般情况下不需要去修改

num.io.threads=8 #45行,用来处理磁盘IO的线程数量,数值应该大于硬盘数

socket.send.buffer.bytes=102400 #48行,发送套接字的缓冲区大小

socket.receive.buffer.bytes=102400 #51行,接收套接字的缓冲区大小

socket.request.max.bytes=104857600 #54行,请求套接字的缓冲区大小

log.dirs=/var/log/kafka #60行,kafka运行日志存放的路径,也是数据存放的路径

num.partitions=1 #65行,topic在当前broker上的默认分区个数,会被topic创建时的指定参数覆盖

num.recovery.threads.per.data.dir=1 #69行,用来恢复和清理data下数据的线程数量

log.retention.hours=168 #103行,segment文件(数据文件)保留的最长时间,单位为小时,默认为7天,超时将被删除

log.segment.bytes=1073741824 #110行,一个segment文件最大的大小,默认为 1G,超出将新建一个新的segment文件

#Kafka 以日志文件的形式维护其数据,而这些日志文件被分割成多个日志段。当一个日志段达到指定的大小时,就会创建一个新的日志段。

配置连接Zookeeper集群地址

zookeeper.connect=192.168.176.70:2181,192.168.176.71:2181,192.168.176.72:2181

kafka默认不允许删除主题

每一台配置一下

修改环境变量日志段是主题分区日志文件的一部分。

vim /etc/profile

export KAFKA_HOME=/opt/kafka

export PATH=$PATH:$KAFKA_HOME/bin

配置 Zookeeper 启动脚本

vim /etc/init.d/kafka

设置开机自启

chmod +x /etc/init.d/kafka

chkconfig --add kafka

分别启动 Kafka

service kafka start

Kafka 命令行操作

创建topic

kafka-topics.sh --create --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --replication-factor 2 --partitions 3 --topic cc

查看当前服务器中的所有 topic

kafka-topics.sh --list --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092

查看某个 topic 的详情

kafka-topics.sh --describe --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092

发布消息

kafka-console-producer.sh --broker-list 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --topic cc

消费消息

kafka-console-consumer.sh --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --topic cc

kafka-console-consumer.sh --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --topic cc --from-beginning

创建单个主题

kafka-topics.sh --create --bootstrap-server 192.168.176.70:9092 --replication-factor 2 --partitions 3 --topic test1

修改分区数

kafka-topics.sh --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --alter --topic test1 --partitions 6

删除 topic

kafka-topics.sh --delete --bootstrap-server 192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092 --topic test1

安装 Filebeat

tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz

mv filebeat-6.7.2-linux-x86_64/ filebeat

安装logstash

#上传软件包 logstash-6.7.2.rpm 到/opt目录下

cd /opt

rpm -ivh logstash-6.7.2.rpm

systemctl start logstash.service

systemctl enable logstash.service

ln -s /usr/share/logstash/bin/logstash /usr/local/bin/

安装nginx或者httpd

yum -y install epel-release

yum -y install nginx

yum -y install httpd

部署 Zookeeper+Kafka 集群

2.部署 Filebeat

cd /usr/local/filebeat

vim filebeat.yml

- type: log

enabled: true

paths:

- /var/log/nginx/access.log

- /var/log/nginx/error.log

tags: ["nginx"]

fields:

service_name: 192.168.176.10_nginx

log_type: nginx

from: 192.168.176.10

output.kafka:

enabled: true

hosts: ["192.168.233.70:9092","192.168.233.71:9092","192.168.233.72:9092"]

topic: "nginx"

底下output 注释

nohup ./filebeat -e -c filebeat.yml >filebeat.out &

tail -f filebeat.out

部署 ELK,在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafka.conf

input {

kafka {

bootstrap_servers => "192.168.176.70:9092,192.168.176.71:9092,192.168.176.72:9092"

topics => "nginx"

type => "nginx_kafka"

codec => "json"

auto_offset_reset => "earliest"

decorate_events => true

}

}

output {

if "nginx" in [tags] {

elasticsearch {

hosts => ["192.168.176.10:9200","192.168.176.50:9200"]

index => "%{[fields][service_name]}-%{+YYYY.MM.dd}"

}

}

stdout { codec => rubydebug }

}

logstash -f kafka.conf --path.data /opt/test1 &

httpd服务

yum -y install httpd

安装 Filebeat

tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz

mv filebeat-6.7.2-linux-x86_64/ filebeat

安装logstash

#上传软件包 logstash-6.7.2.rpm 到/opt目录下

cd /opt

rpm -ivh logstash-6.7.2.rpm

systemctl start logstash.service

systemctl enable logstash.service

ln -s /usr/share/logstash/bin/logstash /usr/local/bin/

部署 Zookeeper+Kafka 集群

2.部署 Filebeat

cd /usr/local/filebeat

vim filebeat.yml

底下output 注释

nohup ./filebeat -e -c filebeat.yml >filebeat.out &

tail -f filebeat.out

在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafkahttpd.conf

logstash -f kafkahttpd.conf --path.data /opt/test1 &

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/207605.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

C++作业3

设计一个Per类&#xff0c;类中包含私有成员:姓名、年龄、指针成员身高、体重&#xff0c;再设计一个Stu类&#xff0c;类中包含私有成员:成绩、Per类对象p1&#xff0c;设计这两个类的构造函数、析构函数和拷贝构造函数。 代码&#xff1a; #include <iostream>using n…

力扣5.最长回文子串

题目描述 思路 1.能够反复利用已判断好的回文子串 2.当子串s[i1,j-1]是回文子串时&#xff0c;只要s[i]s[j]&#xff0c;那么s[i,j]也会是回文子串 3.用好动态规划&#xff0c;具体解释在代码注释里 代码 class Solution {public String longestPalindrome(String s) {int…

【【FPGA 之Micro Blaze的串口中断实验】】

FPGA 之Micro Blaze的串口中断实验 我们在使用 MicroBlaze 进行嵌入式系统设计的时候&#xff0c;通常会用到 AXI Uartlite IP 核与外部设备通信。AXI UART IP 核实现了 RS-232 通讯协议&#xff0c;并使得大家可以设置串口通信相关的波特率、奇偶校验位、停止位和数据位等参数…

一看就懂的RxJava源码分析

一看就懂的RxJava源码分析 前言零、观察模式简介一、RxJava使用示例一二、示例一源码分析0. 示例一代码分解1. RxJava中的观察者是谁&#xff1f;2. RxJava中的被观察者又是谁&#xff1f;3. 观察者又是如何安插到被观察者中的&#xff1f;4. 示例一RxJava源码整体关系类图4. R…

一本书读懂数据治理

企业数据治理非常必要&#xff0c;它是企业实现数字化转型的基础&#xff0c;是企业的一个顶层策略&#xff0c;一个管理体系&#xff0c;也是一个技术体系&#xff0c;涵盖战略、组织、文化、方法、制度、流程、技术和工具等多个层面的内容。 数据治理不是对“数据”的治理&am…

Unity 与 虚拟机ROS连接

Unity 与 虚拟机ROS连接 知识储备前期准备ROS部分Unity部分 连接测试 知识储备 unity官方教程&#xff1a; https://github.com/Unity-Technologies/Unity-Robotics-HubWin11家庭版开启HyperV&#xff1a; https://zhuanlan.zhihu.com/p/577980646HyperV安装Ubuntu: https://b…

SQL Sever 基础知识 - 数据查询

SQL Sever 基础知识 - 一、查询数据 一、查询数据第1节 基本 SQL Server 语句SELECT第2节 SELECT语句示例2.1 SELECT - 检索表示例的某些列2.2 SELECT - 检索表的所有列2.3 SELECT - 对结果集进行筛选2.4 SELECT - 对结果集进行排序2.5 SELECT - 对结果集进行分组2.5 SELECT - …

浅学指针(4)函数指针数组和qsort的使用

系列文章目录 文章目录 系列文章目录前言1.函数指针数组的⽤途作用&#xff1a;可以让代码更简洁&#xff0c;逻辑更清晰 2. 回调函数回调函数就是⼀个通过函数指针调⽤的函数 3 . qsort函数qsort函数可以排序所有数据类型解释如图&#xff1a;![在这里插入图片描述](https://i…

【高效开发工具系列】Hutool DateUtil工具类

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…

Leetcode-二叉树oj题

1.二叉树的前序遍历 144. 二叉树的前序遍历https://leetcode.cn/problems/binary-tree-preorder-traversal/这个题目在遍历的基础上还要求返回数组&#xff0c;数组里面按前序存放二叉树节点的值。 既然要返回数组&#xff0c;就必然要malloc一块空间&#xff0c;那么我们需…

Spring简单的存储和读取

前言 前面讲了spring的创建&#xff0c;现在说说关于Bean和五大类注解 一、Bean是什么&#xff1f; 在 Java 语⾔中对象也叫做 Bean&#xff0c;所以后⾯咱们再遇到对象就以 Bean 著称。这篇文章还是以spring创建为主。 二、存储对象 2.1 俩种存储方式 需要在 spring-conf…

作用域和作用域链

前端面试大全JavaScript作用域和作用域链 &#x1f31f;经典真题 &#x1f31f;作用域&#xff08;Scope&#xff09; 什么是作用域 全局作用域和函数作用域 块级作用域 &#x1f31f;作用域链 什么是自由变量 什么是作用域链 关于自由变量的取值 &#x1f31f;作用域…

初识Linux:权限

目录 提示&#xff1a;以下指令均在Xshell 7 中进行 Linux 的权限 内核&#xff1a; 查看操作系统版本 查看cpu信息 查看内存信息 外部程序&#xff1a; 用户&#xff1a; 普通用户变为超级用户&#xff1a; su 和 su-的区别&#xff1a; root用户变成普通用户&#…

【matlab程序】画海洋流场

【matlab程序】画海洋流场 clear;clc; file ( ‘0227.nc’); latncread(file,‘latitude’); lonncread(file,‘longitude’); uncread(file,‘water_u’); vncread(file,‘water_v’); [x,y]meshgrid(lon,lat); xx’; yy’; interval4; figure (1) set(gcf,‘color’,[1 1 1…

【linux】基本指令(上篇)

1.快速认识5~6个指令 pwd指令 ls指令 touch指令 cd指令 clear指令 touch指令 详细讲解 首先有一个问题就是当我们创建一个文件&#xff0c;但是没有往里面写内容&#xff0c;那么磁盘上会有该文件吗&#xff1f; 磁盘上会保存&#xff0c;因为创建好的文件&#xff0c;没有…

【古月居《ros入门21讲》学习笔记】05_ROS是什么及其核心概念

目录 说明 1. ROS发展史 ROS版本演变 2. ROS是什么 ROS中的通信机制 ROS中的开发工具 ROS中的应用功能 ROS中的生态系统 3. ROS核心概念 节点与节点管理器 通信方式1&#xff1a;话题 通信方式2&#xff1a;服务 话题与服务的区别 参数 文件系统 说明 1. 本系列…

学习笔记7——数据库基础知识以及mysql的查询语句

学习笔记系列开头惯例发布一些寻亲消息 链接&#xff1a;https://baobeihuijia.com/bbhj/contents/3/199913.html 数据库 三个概念区分 DB&#xff1a;数据库&#xff0c;存储数据的仓库&#xff0c;有组织的数据容器DBMS:数据库管理系统SQL&#xff1a;几乎所有的DBMS都支持…

从PDF和图像中提取文本,以供大型语言模型使用

想法 大型语言模型已经席卷了互联网&#xff0c;导致更多的人没有认真关注使用这些模型最重要的部分&#xff1a;高质量的数据&#xff01;本文旨在提供一些有效从任何类型文档中提取文本的技术。 Python库 本文专注于Pytesseract、easyOCR、PyPDF2和LangChain库。实验数据是一…

jQuery的使用

目录 jquery对象&#xff1a; jquery作为一般函数调用参数: jquery事件机制 jquery dom操作 jquery对象&#xff1a; <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta name"viewport" cont…

MySQL 教程 1.4

MySQL 连接 使用mysql二进制方式连接 您可以使用MySQL二进制方式进入到mysql命令提示符下来连接MySQL数据库。 实例 以下是从命令行中连接mysql服务器的简单实例&#xff1a; [roothost]# mysql -u root -p Enter password:****** 在登录成功后会出现 mysql> 命令提示窗…