【Flume】【*】深入flume-ng的三大组件——source,channel,sink

概览

flume-ng中最重要的核心三大组件就是source,channel,sink

source负责从源端收集数据,产出event

channel负责暂存event,以备下游取走消费

sink负责消费通道中的event,写到最终的输出端上

以上是总体的一个简单结构图,下面我们来深入每一个组件的内部看看:

1、Source

source接口的定义如下:


@InterfaceAudience.Public

@InterfaceStability.Stable

public interface Source extends LifecycleAware, NamedComponent {

/**

* Specifies which channel processor will handle this source‘s events.

*

* @param channelProcessor

*/

public void setChannelProcessor(ChannelProcessor channelProcessor);

/**

* Returns the channel processor that will handle this source‘s events.

*/

public ChannelProcessor getChannelProcessor();

}

source生成event并且调用配置的channelprocessor的相关方法,持续的将events存入配置的channel里

channelProcessor中有通道选择器和拦截器链,该过程处在source端收到数据和放入通道直接

该source继承了两个类

1、NamedComponent

负责给每一个组件取一个唯一标识,就是名字,这个名字来源于我们的配置

2、LifecycleAware

负责组件的启停和状态维护

Source接口的直接实现类是AbstractSource抽象类

该类中就定义了通道处理器

还有一个生命状态周期的枚举类型


public enum LifecycleState {

IDLE, START, STOP, ERROR;

public static final LifecycleState[] START_OR_ERROR = new LifecycleState[] {

START, ERROR };

public static final LifecycleState[] STOP_OR_ERROR = new LifecycleState[] {

STOP, ERROR };

}

这里就定义了一个组件会有的4种状态

实现接口的启停组件方法,方法体中只有一个状态的赋值,具体实现,我们来看一个具体的Source——ExecSource

读取配置方面很简单,这里就不说了,看下start方法


public void start() {

logger.info("Exec source starting with command:{}", command);

executor = Executors.newSingleThreadExecutor();

runner = new ExecRunnable(shell, command, getChannelProcessor(), sourceCounter,

restart, restartThrottle, logStderr, bufferCount, batchTimeout, charset);

// FIXME: Use a callback-like executor / future to signal us upon failure.

runnerFuture = executor.submit(runner);

/*

* NB: This comes at the end rather than the beginning of the method because

* it sets our state to running. We want to make sure the executor is alive

* and well first.

*/

sourceCounter.start();

super.start();

logger.debug("Exec source started");

}

该方法内部就是启动了一个线程去执行我们配置的终端命令

前面一篇文章也说过从入口分析如何调用到该start方法,以及调用频率:http://blog.csdn.net/simonchi/article/details/42970373

2、channel

对于通道来说,最重要的就是event的维护,flume的核心就是要中转这些event,所以event一定不能出事

Channel接口定义如下:


@InterfaceAudience.Public

@InterfaceStability.Stable

public interface Channel extends LifecycleAware, NamedComponent {

/**

* <p>Puts the given event into the channel.</p>

* <p><strong>Note</strong>: This method must be invoked within an active

* {@link Transaction} boundary. Failure to do so can lead to unpredictable

* results.</p>

* @param event the event to transport.

* @throws ChannelException in case this operation fails.

* @see org.apache.flume.Transaction#begin()

*/

public void put(Event event) throws ChannelException;

/**

* <p>Returns the next event from the channel if available. If the channel

* does not have any events available, this method must return {@code null}.

* </p>

* <p><strong>Note</strong>: This method must be invoked within an active

* {@link Transaction} boundary. Failure to do so can lead to unpredictable

* results.</p>

* @return the next available event or {@code null} if no events are

* available.

* @throws ChannelException in case this operation fails.

* @see org.apache.flume.Transaction#begin()

*/

public Event take() throws ChannelException;

/**

* @return the transaction instance associated with this channel.

*/

public Transaction getTransaction();

}

通道中的event全部都在事务的管理之中

先来看看这个事务的定义


public interface Transaction {

public enum TransactionState {Started, Committed, RolledBack, Closed };

/**

* <p>Starts a transaction boundary for the current channel operation. If a

* transaction is already in progress, this method will join that transaction

* using reference counting.</p>

* <p><strong>Note</strong>: For every invocation of this method there must

* be a corresponding invocation of {@linkplain #close()} method. Failure

* to ensure this can lead to dangling transactions and unpredictable results.

* </p>

*/

public void begin();

/**

* Indicates that the transaction can be successfully committed. It is

* required that a transaction be in progress when this method is invoked.

*/

public void commit();

/**

* Indicates that the transaction can must be aborted. It is

* required that a transaction be in progress when this method is invoked.

*/

public void rollback();

/**

* <p>Ends a transaction boundary for the current channel operation. If a

* transaction is already in progress, this method will join that transaction

* using reference counting. The transaction is completed only if there

* are no more references left for this transaction.</p>

* <p><strong>Note</strong>: For every invocation of this method there must

* be a corresponding invocation of {@linkplain #begin()} method. Failure

* to ensure this can lead to dangling transactions and unpredictable results.

* </p>

*/

public void close();

}

和我们想想中的一样,也就是一些标准的事务方法的定义,和一个事务状态的枚举类型的定义

基本事务语义抽象类是对它的实现BasicTransactionSemantics

该类定义了两个属性

state状态和initialThreadId,id是唯一的,用来标识事务

构造方法中会赋值为NEW状态,并获取当前事务的一个ID值

重点来看下如下几个方法的具体实现:


 protected void doBegin() throws InterruptedException {}

protected abstract void doPut(Event event) throws InterruptedException;

protected abstract Event doTake() throws InterruptedException;

protected abstract void doCommit() throws InterruptedException;

protected abstract void doRollback() throws InterruptedException;

protected void doClose() {}

1、doBegin

没什么好说的,就是检查状态是否NEW,ID是否匹配,没有问题后,将状态修改为OPEN,表示事务打开

2、doPut

检查ID是否匹配,状态是否打开,event是否为空,为空当然这个put就没意义了

关键看具体是怎么put的?

在FileChannel中有个内部静态类

static class FileBackedTransaction extends BasicTransactionSemantics


 private final LinkedBlockingDeque<FlumeEventPointer> takeList;

private final LinkedBlockingDeque<FlumeEventPointer> putList;

这分别定义了两个双向队列,用于拿和放


protected void doPut(Event event) throws InterruptedException {

channelCounter.incrementEventPutAttemptCount();

if(putList.remainingCapacity() == 0) {

throw new ChannelException("Put queue for FileBackedTransaction " +

"of capacity " + putList.size() + " full, consider " +

"committing more frequently, increasing capacity or " +

"increasing thread count. " + channelNameDescriptor);

}

// this does not need to be in the critical section as it does not

// modify the structure of the log or queue.

if(!queueRemaining.tryAcquire(keepAlive, TimeUnit.SECONDS)) {

throw new ChannelFullException("The channel has reached it‘s capacity. "

+ "This might be the result of a sink on the channel having too "

+ "low of batch size, a downstream system running slower than "

+ "normal, or that the channel capacity is just too low. "

+ channelNameDescriptor);

}

boolean success = false;

log.lockShared();

try {

FlumeEventPointer ptr = log.put(transactionID, event);

Preconditions.checkState(putList.offer(ptr), "putList offer failed "

+ channelNameDescriptor);

queue.addWithoutCommit(ptr, transactionID);

success = true;

} catch (IOException e) {

throw new ChannelException("Put failed due to IO error "

+ channelNameDescriptor, e);

} finally {

log.unlockShared();

if(!success) {

// release slot obtained in the case

// the put fails for any reason

queueRemaining.release();

}

}

}

第一行,跟监控的度量信息有关,表示即将放入通道的event数量+1,监控度量请参考:http://blog.csdn.net/simonchi/article/details/43270461

1、检查队列的剩余空间

2、keepAlive秒时间内获取一个共享信号量,说明put的过程是互斥的

如果该时间内没有成功获取该信号量,那么event放入失败

3、FlumeEventPointer是用来做检查点机制的,因为这是文件通道,会用日志记录的

1、将event和事务id绑定到Pointer上

2、将pointer放入队列尾部

3、通道中的事件队列FlumeEventQueue添加一个未提交的事件,绑定了事务ID

4、释放共享信号量

3、doTake


 protected Event doTake() throws InterruptedException {

channelCounter.incrementEventTakeAttemptCount();

if(takeList.remainingCapacity() == 0) {

throw new ChannelException("Take list for FileBackedTransaction, capacity " +

takeList.size() + " full, consider committing more frequently, " +

"increasing capacity, or increasing thread count. "

+ channelNameDescriptor);

}

log.lockShared();

/*

* 1. Take an event which is in the queue.

* 2. If getting that event does not throw NoopRecordException,

* then return it.

* 3. Else try to retrieve the next event from the queue

* 4. Repeat 2 and 3 until queue is empty or an event is returned.

*/

try {

while (true) {

FlumeEventPointer ptr = queue.removeHead(transactionID);

if (ptr == null) {

return null;

} else {

try {

// first add to takeList so that if write to disk

// fails rollback actually does it‘s work

Preconditions.checkState(takeList.offer(ptr),

"takeList offer failed "

+ channelNameDescriptor);

log.take(transactionID, ptr); // write take to disk

Event event = log.get(ptr);

return event;

} catch (IOException e) {

throw new ChannelException("Take failed due to IO error "

+ channelNameDescriptor, e);

} catch (NoopRecordException e) {

LOG.warn("Corrupt record replaced by File Channel Integrity " +

"tool found. Will retrieve next event", e);

takeList.remove(ptr);

} catch (CorruptEventException ex) {

if (fsyncPerTransaction) {

throw new ChannelException(ex);

}

LOG.warn("Corrupt record found. Event will be " +

"skipped, and next event will be read.", ex);

takeList.remove(ptr);

}

}

}

} finally {

log.unlockShared();

}

}

1、剩余容量检查

2、检查点机制,日志记录操作,从头部取event

3、从takeList中删除该event

4、doCommit


protected void doCommit() throws InterruptedException {

int puts = putList.size();

int takes = takeList.size();

if(puts > 0) {

Preconditions.checkState(takes == 0, "nonzero puts and takes "

+ channelNameDescriptor);

log.lockShared();

try {

log.commitPut(transactionID);

channelCounter.addToEventPutSuccessCount(puts);

synchronized (queue) {

while(!putList.isEmpty()) {

if(!queue.addTail(putList.removeFirst())) {

StringBuilder msg = new StringBuilder();

msg.append("Queue add failed, this shouldn‘t be able to ");

msg.append("happen. A portion of the transaction has been ");

msg.append("added to the queue but the remaining portion ");

msg.append("cannot be added. Those messages will be consumed ");

msg.append("despite this transaction failing. Please report.");

msg.append(channelNameDescriptor);

LOG.error(msg.toString());

Preconditions.checkState(false, msg.toString());

}

}

queue.completeTransaction(transactionID);

}

} catch (IOException e) {

throw new ChannelException("Commit failed due to IO error "

+ channelNameDescriptor, e);

} finally {

log.unlockShared();

}

} else if (takes > 0) {

log.lockShared();

try {

log.commitTake(transactionID);

queue.completeTransaction(transactionID);

channelCounter.addToEventTakeSuccessCount(takes);

} catch (IOException e) {

throw new ChannelException("Commit failed due to IO error "

+ channelNameDescriptor, e);

} finally {

log.unlockShared();

}

queueRemaining.release(takes);

}

putList.clear();

takeList.clear();

channelCounter.setChannelSize(queue.getSize());

}

1、如果putList不为空,提交的是放入通道的事件数量

2、如果takeList不为空,提交的是从通道拿走的事件数量

5、doRollback


  protected void doRollback() throws InterruptedException {

int puts = putList.size();

int takes = takeList.size();

log.lockShared();

try {

if(takes > 0) {

Preconditions.checkState(puts == 0, "nonzero puts and takes "

+ channelNameDescriptor);

synchronized (queue) {

while (!takeList.isEmpty()) {

Preconditions.checkState(queue.addHead(takeList.removeLast()),

"Queue add failed, this shouldn‘t be able to happen "

+ channelNameDescriptor);

}

}

}

putList.clear();

takeList.clear();

queue.completeTransaction(transactionID);

channelCounter.setChannelSize(queue.getSize());

log.rollback(transactionID);

} catch (IOException e) {

throw new ChannelException("Commit failed due to IO error "

+ channelNameDescriptor, e);

} finally {

log.unlockShared();

// since rollback is being called, puts will never make it on

// to the queue and we need to be sure to release the resources

queueRemaining.release(puts);

}

}

在此之前,putList,takeList都没有clear,所以这里可以对着两个双向队列回滚操作

以上是文件通道的实现,如果是内存通道,就没有log的检查点记录了,简单多了,不需要维护状态了。

3、sink

sink的接口定义如下:


@InterfaceAudience.Public

@InterfaceStability.Stable

public interface Sink extends LifecycleAware, NamedComponent {

/**

* <p>Sets the channel the sink will consume from</p>

* @param channel The channel to be polled

*/

public void setChannel(Channel channel);

/**

* @return the channel associated with this sink

*/

public Channel getChannel();

/**

* <p>Requests the sink to attempt to consume data from attached channel</p>

* <p><strong>Note</strong>: This method should be consuming from the channel

* within the bounds of a Transaction. On successful delivery, the transaction

* should be committed, and on failure it should be rolled back.

* @return READY if 1 or more Events were successfully delivered, BACKOFF if

* no data could be retrieved from the channel feeding this sink

* @throws EventDeliveryException In case of any kind of failure to

* deliver data to the next hop destination.

*/

public Status process() throws EventDeliveryException;

public static enum Status {

READY, BACKOFF

}

}

sink与一个通道连接,并消费通道中的events,把它们发送到一个配置的目的地

其实和source的原理大部分相同,同样有一个AbstractSink

我们同样看一个具体实现吧,HDFSEventSink

看它的process方法


public Status process() throws EventDeliveryException {

Channel channel = getChannel();

Transaction transaction = channel.getTransaction();

List<BucketWriter> writers = Lists.newArrayList();

transaction.begin();

…………………………

transaction.commit();

if (txnEventCount < 1) {

return Status.BACKOFF;

} else {

sinkCounter.addToEventDrainSuccessCount(txnEventCount);

return Status.READY;

}

} catch (IOException eIO) {

transaction.rollback();

LOG.warn("HDFS IO error", eIO);

return Status.BACKOFF;

} catch (Throwable th) {

transaction.rollback();

LOG.error("process failed", th);

if (th instanceof Error) {

throw (Error) th;

} else {

throw new EventDeliveryException(th);

}

} finally {

transaction.close();

}

}

这里可以看到,flume-ng在sink端是有事务控制的

事务从 从通道中取event开始,到sink到下一个目的地结束

在这个过程中,任意的失败都会导致事务的回滚,这就保证数据了一致性。

时间: 2024-09-28 03:45:24

【Flume】【*】深入flume-ng的三大组件——source,channel,sink的相关文章

Akka(17): Stream:数据流基础组件-Source,Flow,Sink简介

在大数据程序流行的今天,许多程序都面临着共同的难题:程序输入数据趋于无限大,抵达时间又不确定.一般的解决方法是采用回调函数(callback-function)来实现的,但这样的解决方案很容易造成“回调地狱(callback hell)”,即所谓的“goto-hell”:程序控制跳来跳去很难跟踪,特别是一些变量如果在回调函数中更改后产生不可预料的结果.数据流(stream)是一种解决问题的有效编程方式.Stream是一个抽象概念,能把程序数据输入过程和其它细节隐蔽起来,通过申明方式把数据处理过程

flume的source, channel, sink 列表

Flume Source Source类型 说明 Avro Source 支持Avro协议(实际上是Avro RPC),内置支持 Thrift Source 支持Thrift协议,内置支持 Exec Source 基于Unix的command在标准输出上生产数据 JMS Source 从JMS系统(消息.主题)中读取数据,ActiveMQ已经测试过 Spooling Directory Source 监控指定目录内数据变更 Twitter 1% firehose Source 通过API持续下载

【Flume】从入口Application来分析flume的source和sink是如何与channel交互的

大家在启动flume的时候,输入的命令就可以看出flume的启动入口了 [[email protected] apache-flume-1.5.2-bin]# sh bin/flume-ng agent -c conf -f conf/server.conf -n a1 Info: Sourcing environment configuration script /home/flume/apache-flume-1.5.2-bin/conf/flume-env.sh + exec /home/

【Flume】flume中Avro Source到Avro Sink之间通过SSL传输数据的实现分析及使用

首先你需要了解JAVA KEYSTORE 该SSL用于Avro Sink到Avro Source之间的数据传输 该场景主要用于分布式Flume之间的数据传输,从分散的各个flume agent到中心汇集节点的flume agent 下面来看下如何实现的? Avro Sink SSL 在这个传输过程中,sink其实就相当于socket的client端了 flume源码中有个类NettyAvroRpcClient,该类中还有个内部类SSLCompressionChannelFactory 其中定义了

Flume -- 初识flume、source和sink

Flume – 初识flume.source和sink 目录基本概念常用源 Source常用sink 基本概念 ? 什么叫flume? 分布式,可靠的大量日志收集.聚合和移动工具. ? events 事件,是一行数据的字节数据,是flume发送文件的基本单位. ? flume配置文件 重命名flume-env.sh.template为flume-env.sh,并添加[export JAVA_HOME=/soft/jdk] ? flume的Agent source //从哪儿读数据. 负责监控并收

Flume的Source、Sink总结,及常用使用场景

数据源Source RPC异构流数据交换 Avro Source Thrift Source 文件或目录变化监听 Exec Source Spooling Directory Source Taildir Source MQ或队列订阅数据持续监听 JMS Source SSL and JMS Source Kafka Source Network类数据交换 NetCat TCP Source NetCat UDP Source HTTP Source Syslog Sources Syslog

【Flume】flume中transactionCapacity和batchSize概念的具体分析和解惑

不知道各位用过flume的读者对这两个概念是否熟悉了解 一开始本人的确有点迷惑,觉得这是不是重复了啊? 没感觉到transactionCapacity的作用啊? batchSize又是干啥的啊? -- -- 带着这些问题,我们深入源码来看一下: batchSize batchSize这个概念首先它出现在哪里呢? kafkaSink的process方法 HDFS Sink Exec Source 通过上面这三张图,相信大家应该知道batchSize从哪来的了 batchSize是针对Source和

【Flume】 flume 负载均衡环境的搭建 load_balance

flume的负载均衡即每次按照一定的算法选择sink输出到指定地方,如果在文件输出量很大的情况下,负载均衡还是很有必要的,通过多个通道输出缓解输出压力 flume内置的负载均衡的算法默认是round robin,轮询算法,按序选择 下面看一下具体实例: # Name the components on this agent a1.sources = r1 a1.sinks = k1 k2 a1.channels = c1 # Describe/configure the source a1.so

【Flume】flume 容错环境的搭建 failover

关于failover网上也有很多例子,但是看到的有多重做法,个人觉得,本着职责单一的原则 1.一台机子运行一个flume agent 2.一个agent 的下游sink指向一个flume agent,不要一个flume agent配置多个端口[影响性能] 3.分机子配置,可以避免一台机子司机,另一个仍可以使用,否则陪在一台机子上通过端口区分,一旦死机,全盘崩溃 下面看具体实例: 首先是flumet agent client的配置 priority越高,优先级越高,会优先使用该sink # Nam