Flume日志收集系统架构详解

任何一个生产系统在运行过程中都会产生大量的日志,日志往往隐藏了很多有价值的信息。在没有分析方法之前,这些日志存储一段时间后就会被清理。随着技术的发展和分析能力的提高,日志的价值被重新重视起来。在分析这些日志之前,需要将分散在各个生产系统中的日志收集起来。本节介绍广泛应用的Flume日志收集系统。 一、概述

Flume是Cloudera公司的一款高性能、高可用的分布式日志收集系统,现在已经是Apache的顶级项目。同Flume相似的日志收集系统还有Facebook Scribe、Apache Chuwka。 二、Flume发展历程

Flume 初始的发行版本目前被统称为Flume OG(Original Generation),属于Cloudera。但随着 Flume 功能的扩展,Flume OG 代码工程臃肿、核心组件设计不合理、核心配置不标准等缺点逐渐暴露出来,尤其是在 Flume OG 的最后一个发行版本0.94.0中,日志传输不稳定现象尤为严重。为了解决这些问题,2011 年 10 月 22日,Cloudera 完成了 Flume-728,对Flume进行了里程碑式的改动:重构核心组件、核心配置及代码架构,重构后的版本统称为 Flume NG(Next Generation);改动的另一原因是将 Flume 纳入Apache 旗下,Cloudera Flume 更名为 Apache Flume。 三、Flume架构分析

1. 系统特点

① 可靠性

当节点出现故障时,日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障,从强到弱依次为:end-to-end(收到数据后,Agent首先将事件写到磁盘上,当数据传送成功后,再删除;如果数据发送失败,则重新发送)、Store on Failure(这也是Scribe采用的策略,当数据接收方崩溃时,将数据写到本地,待恢复后继续发送)、Best Effort(数据发送到接收方后,不会进行确认)。

② 可扩展性

Flume采用了三层架构,分别为Agent、Collector和Storage,每一层均可以水平扩展。其中,所有的Agent和Collector均由Master统一管理,这使得系统容易被监控和维护。并且Master允许有多个(使用ZooKeeper进行管理和负载均衡),这样就避免了单点故障问题。

③ 可管理性

当有多个Master时,Flume利用ZooKeeper和Gossip保证动态配置数据的一致性。用户可以在Master上查看各个数据源或者数据流执行情况,并且可以对各个数据源进行配置和动态加载。Flume提供了Web和Shell Script Command两种形式对数据流进行管理。

④ 功能可扩展性

用户可以根据需要添加自己的Agent、Collector或Storage。此外,Flume自带了很多组件,包括各种Agent(如File、Syslog等)、Collector和Storage(如File、HDFS等)。

2. 系统架构

如图所示是Flume OG的架构。

Flume NG的架构如下图所示。Flume采用了分层架构,分别为Agent、Collector和Storage。其中,Agent和Collector均由Source和Sink两部分组成,Source是数据来源,Sink是数据去向。

Flume使用了两个组件:Master和Node。Node根据在Master Shell或Web中的动态配置,决定其是作为Agent还是作为Collector。

① Agent

Agent的作用是将数据源的数据发送给Collector。Flume自带了很多直接可用的数据源(Source),如下。

text("filename"):将文件filename作为数据源,按行发送。

tail("filename"):探测filename新产生的数据,按行发送。

fsyslogTcp(5140):监听TCP的5140端口,并将接收到的数据发送。

tailDir("dirname"[,fileregex=".*"[,startFromEnd=false[,recurseDepth=0]]]):监听目录中的文件末尾,使用正则表达式选定需要监听的文件(不包含目录),recurseDepth为递归监听其下子目录的深度,同时提供了很多Sink,如console[("format")],直接将数据显示在console上。

text("txtfile"):将数据写到文件txtfile中。

dfs("dfsfile"):将数据写到HDFS上的dfsfile文件中。

syslogTcp("host",port):将数据通过TCP传递给host节点。

agentSink[("machine"[,port])]:等价于agentE2ESink,如果省略machine参数,则默认使用flume.collector.event.host与flume.collector.event.port作为默认collectro。

agentDFOSink[("machine"[,port])]:本地热备Agent。Agent发现Collector节点故障后,不断检查Collector的存活状态以便重新发送Event,在此期间产生的数据将缓存到本地磁盘中。

agentBESink[("machine"[,port])]:不负责的Agent。如果Collector出现故障,将不作任何处理,它发送的数据也将被直接丢弃。

agentE2EChain:指定多个Collector,以提高可用性。当向主Collector发送Event失效后,将转向第二个Collector发送;当所有的Collector都失效后,它还会再发送一遍。

② Collector

Collector的作用是将多个Agent的数据汇总后,加载到Storage中。它的Source和Sink与Agent类似。

Source如下。

collectorSource[(port)]:Collector Source,监听端口汇聚数据。

autoCollectorSource:通过Master协调物理节点自动汇聚数据。

logicalSource:逻辑Source,由Master分配端口并监听rpcSink。

Sink如下。

collectorSink("fsdir","fsfileprefix",rollmillis):collectorSink,数据通过Collector汇聚之后发送到HDFS,fsdir是HDFS目录,fsfileprefix为文件前缀码。

customdfs("hdfspath"[,"format"]):自定义格式DFS。

③ Storage

Storage是存储系统,可以是一个普通File,也可以是HDFS、Hive、HBase、分布式存储等。

④ Master

Master负责管理、协调Agent和Collector的配置信息,是Flume集群的控制器。

在Flume中,最重要的抽象是Data Flow(数据流)。Data Flow描述了数据从产生、传输、处理到最终写入目标的一条路径,如下图所示。

对于Agent数据流配置,就是从哪里得到数据,就把数据发送到哪个Collector。

对于Collector,就是接收Agent发送过来的数据,然后把数据发送到指定的目标机器上。

注:Flume框架对Hadoop和ZooKeeper的依赖只存在于JAR包上,并不要求Flume启动时必须将Hadoop和ZooKeeper服务同时启动。

3. 组件介绍

本文所说的Flume基于1.4.0版本。

① Client

路径:apache-flume-1.4.0-src\flume-ng-clients。

操作最初的数据,把数据发送给Agent。在Client与Agent之间建立数据沟通的方式有两种。

第一种方式:创建一个iclient继承Flume已经存在的Source,如AvroSource或者SyslogTcpSource,但是必须保证所传输的数据Source可以理解。

第二种方式:写一个Flume Source通过IPC或者RPC协议直接与已经存在的应用通信,需要转换成Flume可以识别的事件。

Client SDK:是一个基于RPC协议的SDK库,可以通过RPC协议使应用与Flume直接建立连接。可以直接调用SDK的api函数而不用关注底层数据是如何交互的,提供append和appendBatch两个接口,具体的可以看看代码apache-flume-1.4.0-src\flume-ng-sdk\src\main\java\org\apache\ flume\api\RpcClient.java。

② NettyAvroRpcClient

Avro是默认的RPC协议。NettyAvroRpcClient和ThriftRpcClient分别对RpcClient接口进行了实现,具体实现可以看下代码apache-flume-1.4.0-src\flume-ng-sdk\src\main\java\org\apache\flume\api\ NettyAvroRpcClient.java和apache-flume-1.4.0-src\flume-ng-sdk\src\main\java\org\apache\flume\api\ ThriftRpcClient.java。

下面给出一个使用SDK与Flume建立连接的样例如下,实际使用中可以参考实现:

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import java.nio.charset.Charset;
public class MyApp {
public static void main(String[] args) {
MyRpcClientFacade client = new MyRpcClientFacade();
// Initialize client with the remote Flume agent's host and port
client.init("host.example.org",41414);
// Send 10 events to the remote Flume agent. That agent should be
// configured to listen with an AvroSource.
String sampleData = "Hello Flume!";
for (int i = 0; i < 10; i++) {
client.sendDataToFlume(sampleData);
}
client.cleanUp();
}
}
class MyRpcClientFacade {
private RpcClient client;
private String hostname;
private int port;
public void init(String hostname,int port) {
// Setup the RPC connection
this.hostname = hostname;
this.port = port;
this.client = RpcClientFactory.getDefaultInstance(hostname,port);
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname,port);
}
public void sendDataToFlume(String data) {
// Create a Flume Event object that encapsulates the sample data
Event event = EventBuilder.withBody(data,Charset.forName("UTF-8"));
// Send the event
try {
client.append(event);
} catch (EventDeliveryException e) {
// clean up and recreate the client
client.close();
client = null;
client = RpcClientFactory.getDefaultInstance(hostname,port);
// Use the following method to create a thrift client (instead of the above line):
// this.client = RpcClientFactory.getThriftInstance(hostname,port);
}
}
public void cleanUp() {
// Close the RPC connection
client.close();
}
}

为了能够监听到关联端口,需要在配置文件中增加端口和Host配置信息(配置文件apache-flume- 1.4.0-src\conf\flume-conf.properties.template)。

client.type = default (for avro) or thrift (for thrift)
hosts = h1                         # default client accepts only 1 host
                                    # (additional hosts will be ignored)
hosts.h1 = host1.example.org:41414 # host and port must both be specified
                                    # (neither has a default)
batch-size = 100                	# Must be >=1 (default:100)
connect-timeout = 20000        	# Must be >=1000 (default:20000)
request-timeout = 20000        	# Must be >=1000 (default:20000)

除了以上两类实现外,FailoverRpcClient.java和LoadBalancingRpcClient.java也分别对RpcClient接口进行了实现。

③ FailoverRpcClient

该接口主要实现了主备切换,采用<host>:<port>的形式,一旦当前连接失败,就会自动寻找下一个连接。

④ LoadBalancingRpcClient

该接口在有多个Host的时候起到负载均衡的作用。

⑤ Embeded Agent

Flume允许用户在自己的Application里内嵌一个Agent。这个内嵌的Agent是一个轻量级的Agent,不支持所有的Source Sink Channel。

⑥ Transaction

Flume的三个主要组件——Source、Sink、Channel必须使用Transaction来进行消息收发。在Channel的类中会实现Transaction的接口,不管是Source还是Sink,只要连接上Channel,就必须先获取Transaction对象,如下图所示。

具体使用实例如下,可以供生成环境中参考:

Channel ch = new MemoryChannel();
Transaction txn = ch.getTransaction();
txn.begin();
try {
Event eventToStage = EventBuilder.withBody("Hello Flume!",Charset.forName ("UTF-8"));
ch.put(eventToStage);
txn.commit();
} catch (Throwable t) {
txn.rollback();
if (t instanceof Error) {
throw (Error)t;
}
} finally {
txn.close();
}

⑦ Sink

Sink的一个重要作用就是从Channel里获取事件,然后把事件发送给下一个Agent,或者把事件存储到另外的仓库内。一个Sink会关联一个Channel,这是配置在Flume的配置文件里的。SinkRunner.start()函数被调用后,会创建一个线程,该线程负责管理Sink的整个生命周期。Sink需要实现LifecycleAware接口的start()和stop()方法。

Sink.start():初始化Sink,设置Sink的状态,可以进行事件收发。

Sink.stop():进行必要的cleanup动作。

Sink.process():负责具体的事件操作。

Sink使用参考代码实例如下:

public class MySink extends AbstractSink implements Configurable {
private String myProp;
@Override
public void configure(Context context) {
String myProp = context.getString("myProp","defaultValue");
// Process the myProp value (e.g. validation)
// Store myProp for later retrieval by process() method
this.myProp = myProp;
}
@Override
public void start() {
// Initialize the connection to the external repository (e.g. HDFS) that
// this Sink will forward Events to ..
}
@Override
public void stop () {
// Disconnect from the external respository and do any
// additional cleanup (e.g. releasing resources or nulling-out
// field values) ..
}
@Override
public Status process() throws EventDeliveryException {
Status status = null;
// Start transaction
Channel ch = getChannel();
Transaction txn = ch.getTransaction();
txn.begin();
try {
// This try clause includes whatever Channel operations you want to do
Event event = ch.take();
// Send the Event to the external repository.
// storeSomeData(e);
txn.commit();
status = Status.READY;
} catch (Throwable t) {
txn.rollback();
// Log exception,handle individual exceptions as needed
status = Status.BACKOFF;
// re-throw all Errors
if (t instanceof Error) {
throw (Error)t;
}
} finally {
txn.close();
}
return status;
}
}

⑧ Source

Source的作用是从Client端接收事件,然后把事件存储到Channel中。PollableSourceRunner.start()用于创建一个线程,管理PollableSource的生命周期。同样也需要实现start()和stop()两种方法。需要注意的是,还有一类Source,被称为EventDrivenSource。区别是EventDrivenSource有自己的回调函数用于捕捉事件,并不是每个线程都会驱动一个EventDrivenSource。

以下是一个PollableSource的例子:

public class MySource extends AbstractSource implements Configurable, PollableSource {
private String myProp;
@Override
public void configure(Context context) {
String myProp = context.getString("myProp","defaultValue");
// Process the myProp value (e.g. validation,convert to another type,...)
// Store myProp for later retrieval by process() method
this.myProp = myProp;
}
@Override
public void start() {
// Initialize the connection to the external client
}
@Override
public void stop () {
// Disconnect from external client and do any additional cleanup
// (e.g. releasing resources or nulling-out field values) ..
}
@Override
public Status process() throws EventDeliveryException {
Status status = null;
// Start transaction
Channel ch = getChannel();
Transaction txn = ch.getTransaction();
txn.begin();
try {
// This try clause includes whatever Channel operations you want to do
// Receive new data
Event e = getSomeData();
// Store the Event into this Source's associated Channel(s)
getChannelProcessor().processEvent(e)
txn.commit();
status = Status.READY;
} catch (Throwable t) {
txn.rollback();
// Log exception,handle individual exceptions as needed
status = Status.BACKOFF;
// re-throw all Errors
if (t instanceof Error) {
throw (Error)t;
}
} finally {
txn.close();
}
return status;
}
}

4. Flume使用模式

Flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位,它携带日志数据(字节数组形式)并且携带有头信息,这些Event由Agent外部的Source,比如上图中的Web Server生成。当Source捕获事件后会进行特定的格式化,然后Source会把事件推入(单个或多个)Channel中。你可以把Channel看作是一个缓冲区,它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。

很直白的设计,其中值得注意的是,Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source,Channel和Sink可以自由组合。多Agent串联,如下图所示。

或者多Agent合并,如下图所示。

如果你以为Flume就这些能耐那就大错特错了。Flume支持用户建立多级流,也就是说,多个agent可以协同工作,并且支持Fan-in、Fan-out、Contextual Routing、Backup Routes。如下图所示。

参考文献

  • 参考http://www.aboutyun.com/thread-7848-1-1.html官网用户手册,http://flume.apache.org/FlumeUserGuide.html。
  • Github地址https://github.com/apache/flume。
  • 参考http://flume.apache.org/FlumeUserGuide.html。

原文发布于微信公众号 - 大数据和云计算技术(jiezhu2007)

原文发表时间:2017-09-06

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏linux驱动个人学习

基于input子系统的sensor驱动调试(二)

继上一篇:https://cloud.tencent.com/developer/article/1054078 一、驱动流程解析: 1、模块加载: 1 st...

4657
来自专栏积累沉淀

Hive与HBase实现数据互导

建立与HBase的识别表 hive> create table hive_hbase_1(key int,value string)     > stored...

1948
来自专栏杨建荣的学习笔记

巧用闪回数据库来查看历史数据 (r10笔记第47天)

国庆期间有一个例行维护的任务,需要在大早上7点起来,先根据业务指定的SQL查出指定数据,然后运行一个存储过程来更新数据。 查出来的这部分数据需要...

3104
来自专栏杨建荣的学习笔记

生产环境sqlldr加载性能问题及分析之一 (r2第17天)

在测试环境中进行了多轮测试,使用sqlldr批量加载数据,csv文件大概有120G左右,在一致的数据量的情况下,测试环境都在一个小时左右,但是在生产环境中竟然跑...

3296
来自专栏fixzd

redis系列:通过日志案例学习string命令

该文章将通过一个小demo将讲述Redis中的string类型命令。demo将以springboot为后台框架快速开发,iview前端框架进行简单的页面设计,为...

1004
来自专栏SmartSql

SmartCode.ETL 这不是先有鸡还是蛋的问题!

相信不少同学都用过各种代码生成器,这里我就不做详细介绍了,如果想体验 SmartCode.Generator 请至 https://www.cnblogs.co...

1226
来自专栏FreeBuf

Shellcode与加密流量之间的那些事儿

在这篇文章中,我们将简单介绍如何在通过TCP通信的位置无关代码(PIC)中实现数据加密。

702
来自专栏不会写文章的程序员不是好厨师

为Hibiscus写文之定时器篇——HashedWheelTimer

去年一年在简书大约写了25篇,在公司内网写了5篇博客。今年定个小目标吧,在简书产出高质量的博客50篇,加油!

912
来自专栏Felix的技术分享

《一个操作系统的实现》笔记(5)--内核雏形

1254
来自专栏Ceph对象存储方案

查询bucket已用量脚本-python

目前仅支持ceph的s3方案,具体配置看说明 # -*- coding: utf-8 -*- import requests import json from ...

2039

扫码关注云+社区