7.kafka+ELK连接

文章目录

  • kafka+ELK连接
    • 部署Kafka
    • kafka操作命令
    • kafka架构深入
    • Filebeat+Kafka+ELK连接

kafka+ELK连接

部署Kafka

###关闭防火墙

systemctl stop firewalld
systemctl disable firewalld

setenforce 0

vim /etc/selinux/config

SELINUX=disabled

###下载安装包

官方下载地址：http://kafka.apache.org/downloads.html

cd /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz

###安装 Kafka

cd /opt/
tar xf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka

###修改配置文件

##备份配置文件
cd /usr/local/kafka/config/
cp server.properties{,.bak}

vim server.properties

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

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

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

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

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

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

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

---60行---
log.dirs=/usr/local/kafka/logs        
#60行，kafka运行日志存放的路径，也是数据存放的路径

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

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

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

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

---123行---
zookeeper.connect=192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181    
###123行，配置连接Zookeeper集群地址

###修改环境变量

vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin

###加载配置
source /etc/profile


###复制文件到zookeeper服务器哦

scp -r /usr/local/kafka 192.168.242.71:/usr/local
scp -r /usr/local/kafka 192.168.242.72:/usr/local

###部署kafka启动脚本

vim /etc/init.d/kafka


#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
	echo "---------- Kafka 启动 ------------"
	${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
	echo "---------- Kafka 停止 ------------"
	${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
	$0 stop
	$0 start
;;
status)
	echo "---------- Kafka 状态 ------------"
	count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
	if [ "$count" -eq 0 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac

###设置开机自启

cd /etc/init.d/
chmod +x /etc/init.d/kafka
chkconfig --add kafka

###分别启动 Kafka
service kafka start


###另外一种启动方式

cd /usr/local/kafka/bin

./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

netstat -lntp | grep 9092

在这里插入图片描述

kafka操作命令

####创建topic

kafka-topics.sh --create --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --replication-factor 2 --partitions 3 --topic test



--zookeeper：定义 zookeeper 集群服务器地址，如果有多个 IP 地址使用逗号分割，一般使用一个 IP 即可
--replication-factor：定义分区副本数，1 代表单副本，建议为 2 
--partitions：定义分区数 
--topic：定义 topic 名称

在这里插入图片描述

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

kafka-topics.sh --list --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 

###查看某个 topic 的详情

kafka-topics.sh  --describe --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181

在这里插入图片描述

###发布消息

kafka-console-producer.sh --broker-list 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092  --topic test

###消费消息

kafka-console-consumer.sh --bootstrap-server 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092 --topic test --from-beginning

--from-beginning：会把主题中以往所有的数据都读取出来

在这里插入图片描述

###修改分区数

kafka-topics.sh --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181  --alter --topic test --partitions 6

在这里插入图片描述

###删除 topic

kafka-topics.sh --delete --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --topic test

在这里插入图片描述

kafka架构深入

  • Kafka 工作流程及文件存储机制

    • Kafka 中消息是以 topic 进行分类的，生产者生产消息，消费者消费消息，都是面向 topic 的。

  • topic 是逻辑上的概念，而 partition 是物理上的概念，每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。

  • Producer 生产的数据会被不断追加到该 log 文件末端，且每条数据都有自己的 offset。

  • 消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费。

  • 由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。

  • 每个 segment 对应两个文件：“.index” 文件和 “.log” 文件。

  • 这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，test 这个 topic 有三个分区， 则其对应的文件夹为 test-0、test-1、test-2。

  • index 和 log 文件以当前 segment 的第一条消息的 offset 命名。

    • “.index” 文件存储大量的索引信息，
    • “.log” 文件存储大量的数据，
    • 索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

  • 数据可靠性保证

    • 为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后， 都需要向 producer 发送 ack（acknowledgement 确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。

  • 数据一致性问题

    • LEO：指的是每个副本最大的 offset；
    • HW：指的是消费者能见到的最大的 offset，所有副本中最小的 LEO。

  • follower 故障

    • follower 发生故障后会被临时踢出 ISR（Leader 维护的一个和 Leader 保持同步的 Follower 集合），待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

  • leader 故障

    • leader 发生故障之后，会从 ISR 中选出一个新的 leader， 之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据。

  • 注：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

  • ack 应答机制

    • 对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 follower 全部接收成功。
    • 所以 Kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡选择。

  • 当 producer 向 leader 发送数据时，可以通过 request.required.acks 参数来设置数据可靠性的级别：

    • 0：这意味着producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的。当broker故障时有可能丢失数据。
    • 1（默认配置）：这意味着producer在ISR中的leader已成功收到的数据并得到确认后发送下一条message。如果在follower同步成功之前leader故障，那么将会丢失数据。
    • -1（或者是all）：producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。但是如果在 follower 同步完成后，broker 发送ack 之前，leader 发生故障，那么会造成数据重复。

  • 三种机制性能依次递减，数据可靠性依次递增。

注：在 0.11 版本以前的Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。在 0.11 及以后版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据， Server 端都只会持久化一条。

+ kafka会通过ack机制保证数据的可靠性

  • ack配置参数有
  • 0（效果类似异步复制）：不等待follower同步完成就让生产者发送下一条消息
  • 1（效果类似半同步复制）：至少等待一个follower同步完成才让生产者发送下一条消息
  • -1（效果类似全同步复制）：等待所有follower同步完成才让生产者发送下一条消息

Filebeat+Kafka+ELK连接

在这里插入图片描述

###部署 Zookeeper+Kafka 集群

###部署 Filebeat，修改配置文件 

cd /usr/local/filebeat

vim filebeat.yml


filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /var/log/httpd/access_log
  tags: ["access"]


- type: log
  enabled: true
  paths:
    - /var/log/httpd/error_log
  tags: ["error"]
  
......


#添加输出到 Kafka 的配置

output.kafka:
  enabled: true
  hosts: ["192.168.242.70:9092","192.168.242.71:9092","192.168.242.72:9092"]    #指定 Kafka 集群配置
  topic: "httpd"    #指定 Kafka 的 topic

在这里插入图片描述
在这里插入图片描述

###启动 filebeat

./filebeat -e -c filebeat.yml

在这里插入图片描述

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

cd /etc/logstash/conf.d/

vim kafka.conf

input {
    kafka {
        bootstrap_servers => "192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092"  
    #kafka集群地址
    
        topics  => "httpd"     #拉取的kafka的指定topic
        type => "httpd_kafka"  #指定 type 字段
        codec => "json"        #解析json格式的日志数据
        auto_offset_reset => "latest"  #拉取最近数据，earliest为从头开始拉取
        decorate_events => true   #传递给elasticsearch的数据额外增加kafka的属性数据
    }
}

output {
  if "access" in [tags] {
    elasticsearch {
      hosts => ["192.168.242.66:9200"]
      index => "httpd_access-%{+YYYY.MM.dd}"
    }
  }
  
  if "error" in [tags] {
    elasticsearch {
      hosts => ["192.168.242.66:9200"]
      index => "httpd_error-%{+YYYY.MM.dd}"
    }
  }
  
  stdout { codec => rubydebug }
}


####启动 logstash

logstash -f kafka.conf

在这里插入图片描述
在这里插入图片描述

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

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

相关文章

js小写金额转大写 自动转换

js小写金额转大写 自动转换

// 小写转为大写convertCurrency(money) {var cnNums [零, 壹, 贰, 叁, 肆, 伍, 陆, 柒, 捌, 玖]var cnIntRadice [, 拾, 佰, 仟]var cnIntUnits [, 万, 亿, 兆]var cnDecUnits [角, 分, 毫, 厘]// var cnInteger 整var cnIntLast 元var maxNum 999999999999999.9999var…
阅读更多...
船舶中压配电板应用弧光保护，可解决母排故障短路问题，对于提高船舶电站的安全性、稳定性，降低经济损失具有重要意义。-安科瑞黄安南

船舶中压配电板应用弧光保护，可解决母排故障短路问题，对于提高船舶电站的安全性、稳定性，降低经济损失具有重要意义。-安科瑞黄安南

摘要：船舶中压配电板弧光故障导致的设备损坏和停电事故，不仅会造成较大的经济损失，而且严重影响船舶电站的安全稳定运行，威胁船舶电站操作人员的安全。弧光保护是基于电力系统开关柜发生弧光故障时而设计的一套母线保护系统&#…
阅读更多...
使用testify包辅助Go测试指南

使用testify包辅助Go测试指南

我虽然算不上Go标准库的“清教徒”，但在测试方面还多是基于标准库testing包以及go test框架的，除了需要mock的时候，基本上没有用过第三方的Go测试框架。我在《Go语言精进之路》[2]一书中对Go测试组织的讲解也是基于Go testing包和go test框架…
阅读更多...
如何使用MATLAB软件完成生态碳汇涡度相关监测与通量数据分析

如何使用MATLAB软件完成生态碳汇涡度相关监测与通量数据分析

MATLAB MATLAB是美国MathWorks公司出品的商业数学软件，用于数据分析、无线通信、深度学习、图像处理与计算机视觉、信号处理、量化金融与风险管理、机器人，控制系统等领域。 [1] MATLAB是matrix&laboratory两个词的组合，意为矩阵工厂&a…
阅读更多...
力扣206. 反转链表

力扣206. 反转链表

给你单链表的头节点 head ，请你反转链表，并返回反转后的链表。 示例 1： 输入：head [1,2,3,4,5] 输出：[5,4,3,2,1]示例 2： 输入：head [1,2] 输出：[2,1]示例 3： 输入&am…
阅读更多...
stm32(SPI读写W25Q18)

stm32(SPI读写W25Q18)

SPI 是什么？ SPI是串行外设接口（Serial Peripheral Interface）的缩写，是一种高速的，全双工，同步的通信总 线，并且在芯片的管脚上只占用四根线，节约了芯片的管脚，同时为PC…
阅读更多...
electron+vue3全家桶+vite项目搭建【23】url唤醒应用，并传递参数

electron+vue3全家桶+vite项目搭建【23】url唤醒应用，并传递参数

文章目录 引入实现效果实现步骤测试代码 引入 demo项目地址 很多场景下我们都希望通过url快速唤醒应用，例如百度网盘，在网页中唤醒应用，并传递下载链接，在electron中要实现这样的效果，就需要针对不同的平台做对应的处…
阅读更多...
Stable Diffusion Webui 之 ControlNet使用

Stable Diffusion Webui 之 ControlNet使用

一、安装 1.1、插件安装 1.2、模型安装 模型安装分为预处理模型和 controlnet所需要的模型。 先安装预处理模型，打开AI所在的安装目录\extensions\sd-webui-controlnet\annotator,将对应的预处理模型放进对应的文件夹中即可， 而controlnet所需模型则…
阅读更多...
vmware-ubuntu 出现的奇怪问题

vmware-ubuntu 出现的奇怪问题

虚拟机突然连不上网 参考博文-CSDN-卍一十二画卍（作者）-Vmware虚拟机突然连接不上网络【方案集合】 sudo vim /var/lib/NetworkManager/NetworkManager.statesudo service network-manager stop sudo vim /var/lib/NetworkManager/NetworkManager.stat…
阅读更多...
初识react

初识react

初识react 第一步就给我出个问题版本太低 https://www.cnblogs.com/gslgb/p/16585233.html https://blog.csdn.net/xiangshiyufengzhong/article/details/124193898 第二个问题 便利生成dom 需要绑定key 不要总想着加冒号这不是vue 第三个问题 我p标签包裹 MapList组件 MapLis…
阅读更多...
最小二乘拟合平面——直接求解法

最小二乘拟合平面——直接求解法

目录 一、算法原理二、代码实现1、python2、matlab 三、算法效果 一、算法原理 平面方程的一般表达式为： A x B y C z D 0 ( C ≠ 0 ) (1) AxByCzD0(C\neq0)\tag{1} AxByCzD0(C0)(1) 即： z − A C x − B C y − D C (2) z-\frac{A}{C}x-\frac{…
阅读更多...
相机图像质量研究(1)Camera成像流程介绍

相机图像质量研究(1)Camera成像流程介绍

系列文章目录 相机图像质量研究(1)Camera成像流程介绍 相机图像质量研究(2)ISP专用平台调优介绍 相机图像质量研究(3)图像质量测试介绍 相机图像质量研究(4)常见问题总结：光学结构对成像的影响--焦距 相机图像质量研究(5)常见问题总结：光学结构对成…
阅读更多...
Principle Component Analysis

Principle Component Analysis

简述PCA的计算过程 输入：数据集X{x1，x2，...，xn}，需降到k维 ① 去中心化（去均值，即每个特征减去各自的均值） ② 计算协方差矩阵1/nX*X^T（1/n不影响特征向量&#xff09…
阅读更多...
高效出报表的工具有哪些？奥威BI报表工具怎样？

高效出报表的工具有哪些？奥威BI报表工具怎样？

随着企业精细化数据分析的展开，数据分析报表的制作压力也随之增加。对企业而言，拥有一个高效出报表的工具十分重要。高效出报表的工具有哪些？奥威BI报表工具的效率够不够高？ 高效出报表的工具有很多，奥威BI报表工具就…
阅读更多...
mybatis 基础2

mybatis 基础2

查询所有 数据库字段与JavaBean属性保持一致 数据库字段与JavaBean属性不一致 使用resultMap标签 多表查询 association【引入JavaBean实体类】 通过tprice，address_name模糊查询
阅读更多...
「数字化制造」 是如何让制造过程信息化的？

「数字化制造」 是如何让制造过程信息化的？

「数字化制造」 是如何让制造过程信息化的？ 数字化制造是指利用数字技术和信息化手段来实现制造过程的智能化、自动化和高效化。 它通过将传感器、物联网、云计算、大数据分析、人工智能等先进技术与制造业相结合，实现生产过程的数字化、网络化和智能化…
阅读更多...
【stable diffusion】保姆级入门课程-Stable diffusion（SD）介绍与安装

【stable diffusion】保姆级入门课程-Stable diffusion（SD）介绍与安装

目录 0.学前准备 1.什么是AI绘画 2.当前主流的AI绘画工具 3.什么是SD(stable diffusion) 4.SD能做什么 1.文生图 2.图生图 3.AI换模特，背景 5.使用stable diffusion配置要求 6.环境配置与安装 需要注意的地方： 扩展知识： 1.pyth…
阅读更多...
Day57|647. 回文子串 、516.最长回文子序列

Day57|647. 回文子串 、516.最长回文子序列

647. 回文子串 1.题目： 给你一个字符串 s ，请你统计并返回这个字符串中 回文子串 的数目。 回文字符串 是正着读和倒过来读一样的字符串。 子字符串 是字符串中的由连续字符组成的一个序列。 具有不同开始位置或结束位置的子串，即使是…
阅读更多...
日撸java三百行day77-80

日撸java三百行day77-80

文章目录 说明GUI1. GUI 总体布局2. GUI 代码理解2.1 对话框相关控件2.1.1 ApplicationShowdown.java（关闭应用程序）2.1.2 DialogCloser.java（关闭对话框）2.1.3 ErrorDialog.java（显示错误信息）2.1.4 HelpD…
阅读更多...
websoket

websoket

websoket是html5新特性， 它提供一种基于TCP连接上进行全双工通讯的协议; 全双工通信的意思就是:允许客户端给服务器主动发送信息,也支持服务端给另一个客户端发送信息. Websoket使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在we…
阅读更多...
最新文章

Copyright @ 2022~2023