环境准备：

涉及到的技术有flume，Kafka，zookeeper。

操作步骤：

1、构建agent

train.sources=trainSource
train.channels=trainChannel
train.sinks=trainSinktrain.sources.trainSource.type=spooldir
train.sources.trainSource.spoolDir=/opt/kb15tmp/flumelogfile/train
train.sources.trainSource.deserializer=LINE
train.sources.trainSource.deserializer.maxLineLength=320000
train.sources.trainSource.includePattern=train_[0-9]{4}-[0-9]{2}-[0-9]{2}.csv
train.sources.trainSource.interceptors=head_filter
train.sources.trainSource.interceptors.head_filter.type=regex_filter
train.sources.trainSource.interceptors.head_filter.regex=^user*
train.sources.trainSource.interceptors.head_filter.excludeEvents=truetrain.channels.trainChannel.type=file
train.channels.trainChannel.checkpointDir=/opt/kb15tmp/checkpoint/train
train.channels.trainChannel.dataDirs=/opt/kb15tmp/checkpoint/data/traintrain.sinks.trainSink.type=org.apache.flume.sink.kafka.KafkaSink
train.sinks.trainSink.batchSize=640
train.sinks.trainSink.brokerList=192.168.91.180:9092
train.sinks.trainSink.topic=traintrain.sources.trainSource.channels=trainChannel
train.sinks.trainSink.channel=trainChannel

2、启动Kafka和zookeeper
启动zookeeperzkServer.sh start
启动Kafkanohup kafka-server-start.sh /opt/soft/kafka211/config/server.properties &

3、启动消费者进行消费
首先先创建主题，kafka-topics.sh --create --zookeeper 192.168.91.180:2181 --topic train --partitions 1 --replication-factor 1
消费：
kafka-console-consumer.sh --bootstrap-server 192.168.91.180:9092 --topic train --from-beginning

4、启动flume
./bin/flume-ng agent --name train --conf conf/ --conf-file conf/KB15conf/train.conf -Dflume.root.logger=INFO,console

5、将需要消费的日志文件拷贝到指定的文件夹下
cp train.csv /opt/kb15tmp/flumelogfile/train/train_2021-12-27.csv

Flume采集日志到Kafka经典案例相关推荐

Flume采集日志数据
一.为什么选用Flume? Flume vs Logstash vs Filebeat 当时选择数据采集工具时,我们主要参考了市面上热度比较高的Flume和Logstash还有Filebeat,据目前 ...
Kubernetes日志收集：log-pilot采集日志到kafka
1.log-pilot配置环境变量需要修改log-pilot环境变量的两个参数 LOGGING_OUTPUT的值配置为kafka 若将日志采集到kafka,则需要新增KAFKA_BROKERS环境变 ...
SparkStreaming+kafka+flume+hbase日志实时流处理项目
1.项目背景: 互联网访问日志概述为什么要记录访问日志的行为呢? 通过日志我们可以得到网站页面的访问量,网站的黏性,推荐用户行为分析,是指在获得网站访问量基本数据的情况下,对有关数据进行统计.分析, ...
Kafka对接采集日志Flum的集群搭建与部署
目录 Kafka简介消息队列 Kafka的应用场景消息队列的两种模型 Kafka中的重要概念消费者组幂等性 Kafka集群搭建 kafka集群部署 kafka启动脚本 Kafka命令行操作 1 ...
【采集层】Kafka 与 Flume 如何选择
2019独角兽企业重金招聘Python工程师标准>>> 采集层主要可以使用Flume, Kafka两种技术. Flume:Flume 是管道流方式,提供了很多的默认实现,让用户通过 ...
【Linux】logrotate切割Tomcat日志并轮转（Flume采集准备工作适用）
[Linux]logrotate切割Tomcat日志并轮转(Flume采集准备工作适用) 背景及使用场景系统及软件环境解决方案及具体操作 logrotate配置文件编辑 logrotate全部配置 ...
flume采集hive日志写到hdfs问题
解决flume采集hive日志写到hdfs问题在配置flume写日志到hdfs发现写不进去,很苦恼,查了很多资料,终于找到了原因! 原来用Flume监听目录,在自己建的.conf文件中需要配置hd ...
Flume与Kafka整合案例详解
环境配置名称版本下载地址 Centos 7.0 64x 百度 Zookeeper 3.4.5 Flume 1.6.0 Kafka 2.1.0 flume笔记直接贴配置文件 [roo ...
生成器案例，#采集日志
生成器: 只要在函数体中看到yield,那么这个函数就是生成器. yield功能: 1:为我们提供了一种自定义迭代器对象的方法 2:yield其实也是一种返回值,他和return的区别就是:他可以返回 ...

Flume采集日志到Kafka经典案例

环境准备：

操作步骤：

Flume采集日志到Kafka经典案例相关推荐

最新文章

热门文章