[flume] channel 和 sink

由于经常会使用到Flume的一些channel,source,sink,于是为了方便将这些channel,source,sink汇总出来,也共大家访问. Component Interface Type Alias Implementation Class *.Channel memory *.channel.MemoryChannel *.Channel jdbc *.channel.jdbc.JdbcChannel *.Channel file *.channel.file.FileChan

说明: 该结果是亲自测试,只提供简单的数据分析,很简陋,结果可能不准确. 先说一下结果,多sink可以直接按常规配置,这样的话每个sink会启动一个sinkrunner,相当于每个线程一个sink,互不干扰,负载均衡是通过channel实现的,效率会提高为n倍,如果在此基础上加入 sinkgroup,则sinkgroup会启动一个sinkrunner,就是单线程,sinkgroup从channel中读取数据,然后分发到下面挂载的sink中,效率和单sink一样,没有提高,但是可以实现两个sink

概览 flume-ng中最重要的核心三大组件就是source,channel,sink source负责从源端收集数据,产出event channel负责暂存event,以备下游取走消费 sink负责消费通道中的event,写到最终的输出端上 以上是总体的一个简单结构图,下面我们来深入每一个组件的内部看看: 1.Source source接口的定义如下: @InterfaceAudience.Public @InterfaceStability.Stable public interface S

import java.io.File; import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import org.apache.flume.Channel; import org.apache.flume.Context; import org.apache.flume.Event; import org.apache.flume.EventDeli

对于flume的监控,只需要监控channel的性能数据即可,source和sink的性能一部分可以从channel中表现出来.以MemoryChannel为例,在MemoryTransaction的构造函数中会实例化一个org.apache.flume.instrumentation.ChannelCounter对象     public MemoryTransaction( int transCapacity, ChannelCounter counter) {       putList 

参考(向原作者致敬) http://ydt619.blog.51cto.com/316163/1230586 https://blogs.apache.org/flume/entry/streaming_data_into_apache_hbase flume 1.5 的配置文件示例 #Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the

如图所示,新建一个JAVA工程,编辑pom文件,pom文件内容如下[这里取出了parent]: <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLoca

一.概述: 在实际的生产环境中,一般都会遇到将web服务器比如tomcat.Apache等中产生的日志倒入到HDFS中供分析使用的需求.这里的配置方式就是实现上述需求. 二.配置文件: #agent1 name agent1.sources=source1 agent1.sinks=sink1 agent1.channels=channel1 #Spooling Directory #set source1 agent1.sources.source1.type=spooldir agent1.

a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = netcat a1.sources.r1.bind = localhost a1.sources.r1.port = 44444 # Describe the sink #a1.sinks.k1.type = logger a1.sinks.k1.type = hdfs a1.sinks.k1.hd