Flume集群搭建

Flume集群搭建 一：Flume简介

Flume是一个分布式的、高可用的海量日志收集、聚合和传输日志收集系统，支持在日志系统中定制各类数据发送方（如：Kafka，HDFS等），便于收集数据。

Flume提供了丰富的日志源收集类型，有：Console、RPC、Text、Tail、Syslog、Exec等数据源的收集。

Agent层：

对于Agent这一层来说，每个机器部署一个Agent，可以水平扩展，不受限制。一个方面，Agent收集日志的能力受限于机器的性能，正常情况下一个Agent可以为单机提供足够服务。另一方面，如果机器比较多，可能受限于后端Collector提供的服务，但Agent到Collector是有Load Balance机制，使得Collector可以线性扩展提高能力。

Collector层：

对于Collector这一层，Agent到Collector是有Load Balance机制，并且Collector提供无差别服务，所以可以线性扩展。其性能主要受限于Store层提供的能力。

Store层：

对于Store这一层来说，Hdfs和Kafka都是分布式系统，可以做到线性扩展。Bypass属于临时的应用，只对应于某一类日志，性能不是瓶颈。

此处第一个测试的日志收集源，使用spooldir方式进行日志文件采集，数据发送方则是HDFS；

第二个测试的日志收集源同样是spooldir的方式采集，而数据发送方则是Kafka。

二、环境准备 a) Hadoop的HDFS，下载并解压flume-1.6.0（3台机器都需要安装）。 b) 规划 Hostname master worker1 worker2 作用 agent collector collector 方式 spooldir HDFS HDFS 路径 /usr/local/flume/tmp/TestDir /library/flume /library/flume

三、配置（source：local file system；sink：HDFS）

a) 系统环境变量（3台机器一样）

添加FLUME_HOME以及PATH中添加${FLUME_HOME}/bin:

b) conf/flume-env.sh配置 添加对JDK的支持即可：

c) master节点上配置/conf/flume-client.properties(source的日志收集) #agent1 name

agent1.channels = c1 agent1.sources = r1 agent1.sinks = k1 k2

#set gruop

agent1.sinkgroups = g1

#set channel

agent1.channels.c1.type = memory agent1.channels.c1.capacity = 1000

agent1.channels.c1.transactionCapacity = 100

agent1.sources.r1.channels = c1 agent1.sources.r1.type = spooldir

agent1.sources.r1.spoolDir =/usr/local/flume/tmp/TestDir

agent1.sources.r1.interceptors = i1 i2

agent1.sources.r1.interceptors.i1.type = static agent1.sources.r1.interceptors.i1.key = Type agent1.sources.r1.interceptors.i1.value = LOGIN agent1.sources.r1.interceptors.i2.type = timestamp

# set sink1

agent1.sinks.k1.channel = c1 agent1.sinks.k1.type = avro

agent1.sinks.k1.hostname = worker1 agent1.sinks.k1.port = 52020

# set sink2

agent1.sinks.k2.channel = c1 agent1.sinks.k2.type = avro

agent1.sinks.k2.hostname = worker2 agent1.sinks.k2.port = 52020

#set sink group

agent1.sinkgroups.g1.sinks = k1 k2

#set failover

agent1.sinkgroups.g1.processor.type = failover agent1.sinkgroups.g1.processor.priority.k1 = 10 agent1.sinkgroups.g1.processor.priority.k2 = 1

agent1.sinkgroups.g1.processor.maxpenalty = 10000

c) worker1、worker2节点上配置/conf/flume-servre.properties worker1上： #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.transactionCapacity = 100

# othernode,nna to nns a1.sources.r1.type = avro a1.sources.r1.bind = worker1 a1.sources.r1.port = 52020

a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = static a1.sources.r1.interceptors.i1.key = Collector a1.sources.r1.interceptors.i1.value = worker1 a1.sources.r1.channels = c1

#set sink to hdfs

a1.sinks.k1.type=hdfs

a1.sinks.k1.hdfs.path=/library/flume a1.sinks.k1.hdfs.fileType=DataStream a1.sinks.k1.hdfs.writeFormat=TEXT a1.sinks.k1.hdfs.rollInterval=1 a1.sinks.k1.channel=c1

a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

worker2上： #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.transactionCapacity = 100

# othernode,nna to nns a1.sources.r1.type = avro a1.sources.r1.bind = worker2 a1.sources.r1.port = 52020 a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = static a1.sources.r1.interceptors.i1.key = Collector a1.sources.r1.interceptors.i1.value = worker2 a1.sources.r1.channels = c1 #set sink to hdfs

a1.sinks.k1.type=hdfs

a1.sinks.k1.hdfs.path=/library/flume a1.sinks.k1.hdfs.fileType=DataStream a1.sinks.k1.hdfs.writeFormat=TEXT a1.sinks.k1.hdfs.rollInterval=1 a1.sinks.k1.channel=c1

a1.sinks.k1.hdfs.filePrefix=%Y-%m-%d

四、启动Flume集群

a) 先启动collector端，即worker1和worker2上的配置文件：

集群启动后，查看各自的进程:Flume的进程名字叫做“Application”

b) 再启动agent端，即master的配置文件：

五、测试数据的传输

在master节点指定的路径下，生成新的文件（模拟web Server产生新的日志文件）： 首先，TestDir下没有文件：

而HDFS指定的路径下也没有文件：

之后，copy2个日志文件到此目录下：

此时，我们看worker1上的控制台打印的日志如下：

最后，我们查看HDFS上的生成的文件的内容：

我们看到，源端的2个文件合并到HDFS上的1个文件中了。

本文来源：https://www.bwwdw.com/article/aq3f.html