flume 整合kafka

背景:系统的数据量越来越大,日志不能再简单的文件的保存,如此日志将会越来越大,也不方便查找与分析,综合考虑下使用了flume来收集日志,收集日志后向kafka传递消息,下面给出具体的配置

# The configuration file needs to define the sources,
# the channels and the sinks.
# Sources, channels and sinks are defined per agent,
# in this case called ‘agent‘

agent.sources = r1
agent.channels = c1
agent.sinks = s1

# For each one of the sources, the type is defined
agent.sources.r1.type = netcat
agent.sources.r1.bind = localhost
agent.sources.r1.port = 10245
agent.sources.r1.charset = UTF-8

# The channel can be defined as follows.
agent.sources.r1.channels = c1

# Each sink‘s type must be defined
agent.sinks.s1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.s1.topic = test
agent.sinks.s1.brokerList = ip:9092
agent.sinks.s1.requiredAcks = 1
agent.sinks.s1.batchSize = 20
agent.sinks.s1.channel = c1

# Each channel‘s type is defined.
agent.channels.c1.type = memory

# Other config values specific to each type of channel(sink or source)
# can be defined as well
# In this case, it specifies the capacity of the memory channel
agent.channels.c1.capacity = 100

启动方式:

   bin/flume-ng agent --conf conf --conf-file conf/kafka.conf --name agent -Dflume.root.logger=INFO,console

再启动之前一定要先启动kafka,这里可能会有一个错误

  

Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch Expired

这个是因为默认情况下kafka是广播的localhost,所以如果不是同一个机器需要修改下配置

advertised.listeners=PLAINTEXT://ip:9092把默认的localhost替换成IP地址 重新启动下就可以了.

时间: 2024-09-30 19:04:47

flume 整合kafka的相关文章

flume 整合 kafka

flume 整合 kafka: flume 采集业务日志,发送到kafka 安装部署Kafka Download 1.0.0 is the latest release. The current stable version is 1.0.0. You can verify your download by following these procedures and using these KEYS. 1.0.0 Released November 1, 2017 Source downloa

Flume整合Kafka完成实时数据采集

agent选择 agent1 exec source + memory channel + avro sink agent2 avro source + memory channel 模拟实际工作中的场景,agent1 为A机器,agent2 为B机器. avro source: 监听avro端口,并且接收来自外部avro信息, avro sink:一般用于跨节点传输,主要绑定数据移动目的地的ip和port 在创建agent2配置文件 cd /app/flume/flume/conf vi te

flume整合kafka

一.需求 利用flume采集Linux下的文件信息,并且传入到kafka集群当中. 环境准备zookeeper集群和kafka集群安装好. 二.配置flume 官网下载flume.博主自己这里使用的是flume1.6.0. 官网地址http://flume.apache.org/download.html 解压缩.tar -zxvf apache-flume-1.6.0-bin.tar.gz -C /usr/apps/ 创建flume配置文件. cd /usr/apps/flume/apache

Flume 学习笔记之 Flume NG+Kafka整合

Flume NG集群+Kafka集群整合: 修改Flume配置文件(flume-kafka-server.conf),让Sink连上Kafka hadoop1: #set Agent name a1.sources = r1 a1.channels = c1 a1.sinks = k1 #set channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacit

ambari下的flume和kafka整合

1.配置flume 1 #扫描指定文件配置 2 agent.sources = s1 3 agent.channels = c1 4 agent.sinks = k1 5 6 agent.sources.s1.type=exec 7 agent.sources.s1.command=tail -F /home/flume/test/test.log 8 agent.sources.s1.channels=c1 9 agent.channels.c1.type=memory 10 agent.ch

2016年大数据Spark“蘑菇云”行动之flume整合spark streaming

近期,听了王家林老师的2016年大数据Spark"蘑菇云"行动,需要将flume,kafka和Spark streaming进行整合. 感觉一时难以上手,还是先从简单着手吧:我的思路是这样的,flume产生数据,然后输出到spark streaming,flume的源数据是netcat(地址:localhost,端口22222),输出是avro(地址:localhost,端口是11111).Spark streaming的处理是直接输出有几个events. 一.配置文件 Flume 配

storm 整合 kafka之保存MySQL数据库

整合Kafka+Storm,消息通过各种方式进入到Kafka消息中间件,比如通过使用Flume来收集的日志数据,然后暂由Kafka中的路由暂存,然后在由实时计算程序Storm做实时分析,这时候我们需要讲Storm中的Spout中读取Kafka中的消息,然后交由具体的Bolt组件分析处理.实际上在 apache-storm-0.9.3这个版本的Storm已经自带了一个集成Kafka的外部插件程序storm-kafka,可以直接使用,例如我使用的Maven依赖配置. 1.配置Maven依赖包 [ht

Flume 与Kafka区别

今天开会讨论日志处理为什么要同时使用Flume和Kafka,是否可以只用Kafka 不使用Flume?当时想到的就只用Flume的接口多,不管是输入接口(socket 和 文件)以及输出接口(Kafka/HDFS/HBase等). 考虑单一应用场景,从简化系统的角度考虑,在满足应用需求的情况下可能只使用一个比较好.但是考虑到现有系统业务发展,为了后面的灵活扩展,在先用系统设计时留有一定的扩展性感觉更重要,可能使用Flume+kafka架构相对只使用Kafka会多占用1-2台机器做Flume日志采

消息系统Flume与Kafka的区别

首先Flume和Kafka都是消息系统,但是它俩也有着很多不同的地方,Flume更趋向于消息采集系统,而Kafka更趋向于消息缓存系统. [一]设计上的不同 Flume是消息采集系统,它主要解决问题是消息的多元采集.因此Flume在实现上提供了多达十几种的Flume Source,以供用户根据不同的应用场景来采集数据.也正因为Flume提供了这 些采集消息的Flume Source,使得用户采集消息变得很简单,用户往往只需要对原始数据稍作处理然后将数据发送给Flume Source.在Flume