日志采集传输框架之 Flume,将监听端口数据发送至Kafka

2024-01-16 00:52

本文主要是介绍日志采集传输框架之 Flume,将监听端口数据发送至Kafka,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1、简介        

        Flume 是 Cloudera 提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传
输的系统。Flume 基于流式架构,主要有以下几个部分组成。

 主要组件介绍:

1)、Flume Agent 是一个 JVM 进程,它以事件的形式将数据从源头送至目的。Agent 主要有 3 个部分组成, Source 、 Channel 、 Sink 。

2)、Source 是负责接收数据到 Flume Agent 的组件。 Source 组件可以处理各种类型、各种
格式的日志数据,包括 avro 、 thrift 、 exec 、 jms 、netcat 、 taildir 、syslog 、 http 。

3)、Sink 不断地轮询 Channel 中的事件且批量地移除它们,并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent 。Sink 组件目的地包括 hdfs、logger、avro 、 thrift、file 、 HBase 、 solr 、自定义。

4)、Channel 是位于 Source 和 Sink 之间的缓冲区。因此, Channel 允许 Source 和 Sink 运作在不同的速率上。 Channel 是线程安全的,可以同时处理几个 Source 的写入操作和几个Sink 的读取操作。Flume 自带两种 Channel Memory Channel File Channel
        Memory Channel:是内存中的队列。 Memory Channel 在不需要关心数据丢失的情景下适用。如果需要关心数据丢失,那么 Memory Channel 就不应该使用,因为程序死亡、机器宕机或者重启都会导致数据丢失。
        File Channel:将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。

5)、Event 传输单元,Flume 数据传输的基本单元,以 Event 的形式将数据从源头送至目的地。Event 由 Header 和 Body 两部分组成, Header 用来存放该 event 的一些属性,为 K-V 结构、Body 用来存放该条数据,形式为字节数组。

2、下载安装

下载地址:Download — Apache Flume

本文下载安装包为:apache-flume-1.11.0-bin.tar.gz

解压安装

# 解压到指定目录
tar -zxvf apache-flume-1.11.0-bin.tar.gz -C /opt/software
 3、案例        

        使用监听端口数据,发送到kafka中。Kafka 集群搭建启动参考博客:kafka丢弃zookeeper,使用kraft架构-CSDN博客

3.1、创建Flume agent 配置文件 vim flume-netcat-kafka.conf:
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 192.168.30.90
a1.sources.r1.port = 44444# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = topic1
a1.sinks.k1.kafka.bootstrap.servers = 192.168.30.88:9092,192.168.30.89:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
3.2、安装 netcat 工具
# 安装 netcat 工具
yum install -y nc
3.3、启动flume
# 1、第一种方式
bin/flume-ng agent --conf conf/ --name a1 --conf-file test/flume-netcat-kafka.conf -
Dflume.root.logger=INFO,console
# 2、第二种方式
bin/flume-ng agent -c conf/ -n a1 -f test/flume-netcat-kafka.conf -Dflume.root.logger=INFO,console

参数说明:

1)、--conf/-c:表示配置文件存储在conf/目录;
2)、--name/-n:表示给agent 起名为a1;
3)、--conf-file/-f:flume 本次启动读取的配置文件是在 test 文件夹下的 flume-netcat-kafka.conf 文件。
4)、-Dflume.root.logger=INFO,console :-D 表示flume 运行时动态修改 flume.root.logger 参数属性值,并将控制台日志打印级别设置为INFO 级别。日志级别包括:log、info、warn、error。

3.4、测试
3.4.1、启动监听44444端口
nc 192.168.30.90 44444
3.4.2、启动 Kafka 消费者控制台
bin/kafka-console-consumer.sh --bootstrap-server 192.168.30.88:9092 --topic topic1 --from-beginning
3.4.3、结果展示

4、总结 

        本文详细介绍 flume 组件各个模块的含义,并且实现从监听端口数据发送至Kafka案例,帮助大家快速掌握flume的使用。关于flume 如何自定义 Source 和 Sink,将在后续博客中更新。

        本人是一个从小白自学计算机技术,对运维、后端、各种中间件技术、大数据等有一定的学习心得,想获取自学总结资料(pdf版本)或者希望共同学习,关注微信公众号:it自学社团。后台回复相应技术名称/技术点即可获得。(本人学习宗旨:学会了就要免费分享)

这篇关于日志采集传输框架之 Flume,将监听端口数据发送至Kafka的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/610780

相关文章

批量导入txt数据到的redis过程

《批量导入txt数据到的redis过程》用户通过将Redis命令逐行写入txt文件,利用管道模式运行客户端,成功执行批量删除以Product*匹配的Key操作,提高了数据清理效率... 目录批量导入txt数据到Redisjs把redis命令按一条 一行写到txt中管道命令运行redis客户端成功了批量删除k

Spring Boot集成/输出/日志级别控制/持久化开发实践

《SpringBoot集成/输出/日志级别控制/持久化开发实践》SpringBoot默认集成Logback,支持灵活日志级别配置(INFO/DEBUG等),输出包含时间戳、级别、类名等信息,并可通过... 目录一、日志概述1.1、Spring Boot日志简介1.2、日志框架与默认配置1.3、日志的核心作用

SpringBoot多环境配置数据读取方式

《SpringBoot多环境配置数据读取方式》SpringBoot通过环境隔离机制,支持properties/yaml/yml多格式配置,结合@Value、Environment和@Configura... 目录一、多环境配置的核心思路二、3种配置文件格式详解2.1 properties格式(传统格式)1.

解决pandas无法读取csv文件数据的问题

《解决pandas无法读取csv文件数据的问题》本文讲述作者用Pandas读取CSV文件时因参数设置不当导致数据错位,通过调整delimiter和on_bad_lines参数最终解决问题,并强调正确参... 目录一、前言二、问题复现1. 问题2. 通过 on_bad_lines=‘warn’ 跳过异常数据3

深度解析Nginx日志分析与499状态码问题解决

《深度解析Nginx日志分析与499状态码问题解决》在Web服务器运维和性能优化过程中,Nginx日志是排查问题的重要依据,本文将围绕Nginx日志分析、499状态码的成因、排查方法及解决方案展开讨论... 目录前言1. Nginx日志基础1.1 Nginx日志存放位置1.2 Nginx日志格式2. 499

kkFileView启动报错:报错2003端口占用的问题及解决

《kkFileView启动报错:报错2003端口占用的问题及解决》kkFileView启动报错因office组件2003端口未关闭,解决:查杀占用端口的进程,终止Java进程,使用shutdown.s... 目录原因解决总结kkFileViewjavascript启动报错启动office组件失败,请检查of

Python Web框架Flask、Streamlit、FastAPI示例详解

《PythonWeb框架Flask、Streamlit、FastAPI示例详解》本文对比分析了Flask、Streamlit和FastAPI三大PythonWeb框架:Flask轻量灵活适合传统应用... 目录概述Flask详解Flask简介安装和基础配置核心概念路由和视图模板系统数据库集成实际示例Stre

C#监听txt文档获取新数据方式

《C#监听txt文档获取新数据方式》文章介绍通过监听txt文件获取最新数据,并实现开机自启动、禁用窗口关闭按钮、阻止Ctrl+C中断及防止程序退出等功能,代码整合于主函数中,供参考学习... 目录前言一、监听txt文档增加数据二、其他功能1. 设置开机自启动2. 禁止控制台窗口关闭按钮3. 阻止Ctrl +

java如何实现高并发场景下三级缓存的数据一致性

《java如何实现高并发场景下三级缓存的数据一致性》这篇文章主要为大家详细介绍了java如何实现高并发场景下三级缓存的数据一致性,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 下面代码是一个使用Java和Redisson实现的三级缓存服务,主要功能包括:1.缓存结构:本地缓存:使

在MySQL中实现冷热数据分离的方法及使用场景底层原理解析

《在MySQL中实现冷热数据分离的方法及使用场景底层原理解析》MySQL冷热数据分离通过分表/分区策略、数据归档和索引优化,将频繁访问的热数据与冷数据分开存储,提升查询效率并降低存储成本,适用于高并发... 目录实现冷热数据分离1. 分表策略2. 使用分区表3. 数据归档与迁移在mysql中实现冷热数据分