`
qianshangding
  • 浏览: 128260 次
  • 性别: Icon_minigender_1
  • 来自: 深圳
社区版块
存档分类
最新评论

Flume之ChannelSelector源码分析

 
阅读更多

Flume中channel选择器(selector.type配置)必须实现ChannelSelector接口,实现了该接口的类主要作用是告诉Source中接收到的Event应该发送到哪些Channel,在Flume中主要由两个实现方式:

1,复用,实现类:MultiplexingChannelSelector

2,复制,实现类:ReplicatingChannelSelector

如下:

ChannelSelector接口两个主要的方法是:

//获取必选的Channel列表
public List<Channel> getRequiredChannels(Event event);
//获取可选的Channel列表
public List<Channel> getOptionalChannels(Event event);

ReplicatingChannelSelector (所有Channel默认的方式)

属性名默认描述
selector.type replicating 组件名:replicating
selector.optional 标记哪些Channels是可选的

以下例子将c3标记为可选,写入c3失败的话会被忽略,如果写入c1和c2失败的话,这个事务就会失败:

a1.sources = r1
a1.channels = c1 c2 c3
a1.source.r1.selector.type = replicating
a1.source.r1.channels = c1 c2 c3
a1.source.r1.selector.optional = c3

ReplicatingChannelSelector初始化过程:
public void configure(Context context) {
    //获取哪些Channel标记为可选
    String optionalList = context.getString(CONFIG_OPTIONAL);
    //将所有Channel都方法必须的Channel列表中
    requiredChannels = new ArrayList<Channel>(getAllChannels());
    Map<String, Channel> channelNameMap = getChannelNameMap();
    if(optionalList != null && !optionalList.isEmpty()) {
      //下面的操作:如果channel属于可选的,则加入可选的列表中,并从必选的列表中删除
      for(String optional : optionalList.split("\\s+")) {
        Channel optionalChannel = channelNameMap.get(optional);
        requiredChannels.remove(optionalChannel);
        if (!optionalChannels.contains(optionalChannel)) {
          optionalChannels.add(optionalChannel);
        }
      }
    }
  }

MultiplexingChannelSelector

属性名默认Description
selector.type replicating 组件名:multiplexing
selector.optional
标记哪些Channels是可选的
selector.header
flume.selector.header

selector.default

selector.mapping.*

示例:

a1.sources = r1
a1.channels = c1 c2 c3 c4
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = state
a1.sources.r1.selector.mapping.CZ = c1
a1.sources.r1.selector.mapping.US = c2 c3
a1.sources.r1.selector.default = c4
根据header中key为state的值,决定将数据写入那个channel中,如上示例将state=CZ写入到c1中,将state=US写入到c2,c3中,默认情况下写入c4

MultiplexingChannelSelector的初始化过程:
  public void configure(Context context) {
    //获取Header的值
    this.headerName = context.getString(CONFIG_MULTIPLEX_HEADER_NAME,
        DEFAULT_MULTIPLEX_HEADER);

    Map<String, Channel> channelNameMap = getChannelNameMap();
    //获取默认的Channel
    defaultChannels = getChannelListFromNames(
        context.getString(CONFIG_DEFAULT_CHANNEL), channelNameMap);

    //获取Mapping的值
    Map<String, String> mapConfig =
        context.getSubProperties(CONFIG_PREFIX_MAPPING);

    //channelMapping变量存放了header变量中必须的Channel列表
    channelMapping = new HashMap<String, List<Channel>>();

    //将header对应的Channels存放到channelMapping变量中。
    for (String headerValue : mapConfig.keySet()) {
      List<Channel> configuredChannels = getChannelListFromNames(
          mapConfig.get(headerValue),
          channelNameMap);

      //This should not go to default channel(s)
      //because this seems to be a bad way to configure.
      if (configuredChannels.size() == 0) {
        throw new FlumeException("No channel configured for when "
            + "header value is: " + headerValue);
      }

      if (channelMapping.put(headerValue, configuredChannels) != null) {
        throw new FlumeException("Selector channel configured twice");
      }
    }
    //If no mapping is configured, it is ok.
    //All events will go to the default channel(s).
    Map<String, String> optionalChannelsMapping =
        context.getSubProperties(CONFIG_PREFIX_OPTIONAL + ".");
    //以下这一整段代码中是赛选出Header对应那些可选Channel列表。
    optionalChannels = new HashMap<String, List<Channel>>();
    for (String hdr : optionalChannelsMapping.keySet()) {
      List<Channel> confChannels = getChannelListFromNames(
              optionalChannelsMapping.get(hdr), channelNameMap);
      if (confChannels.isEmpty()) {
        confChannels = EMPTY_LIST;
      }
      //Remove channels from optional channels, which are already
      //configured to be required channels.

      List<Channel> reqdChannels = channelMapping.get(hdr);
      //Check if there are required channels, else defaults to default channels
      if(reqdChannels == null || reqdChannels.isEmpty()) {
        //如果header对应的必选Channel列表为空,那么deault就作为它的必选Channel
        reqdChannels = defaultChannels;
      }
      for (Channel c : reqdChannels) {
        //如果header对应的Channel是必选的,那么就在可选的列表中删除。
        if (confChannels.contains(c)) {
          confChannels.remove(c);
        }
      }

      if (optionalChannels.put(hdr, confChannels) != null) {
        throw new FlumeException("Selector channel configured twice");
      }
    }

  }

在看看MultiplexingChannelSelector中getRequiredChannels和getOptionalChannels方法,这两个方法也是根据HeaderName来获取Channel列表的:
  @Override
  public List<Channel> getRequiredChannels(Event event) {
    String headerValue = event.getHeaders().get(headerName);
    //headerValue不存在,就获取默认
    if (headerValue == null || headerValue.trim().length() == 0) {
      return defaultChannels;
    }
    //根据headerName获取必选的Channel列表
    List<Channel> channels = channelMapping.get(headerValue);

    //This header value does not point to anything
    //Return default channel(s) here.
    //必选列表为null,则返回默认的Channel列表
    if (channels == null) {
      channels = defaultChannels;
    }

    return channels;
  }

  @Override
  public List<Channel> getOptionalChannels(Event event) {
    String hdr = event.getHeaders().get(headerName);
    //根据HeaderValue获取可选的channel列表
    List<Channel> channels = optionalChannels.get(hdr);
    //可选列表为null,则返回空列表。
    if(channels == null) {
      channels = EMPTY_LIST;
    }
    return channels;
  }

自定义 Channel Selector

自定义的Channel Selector必须实现ChannelSelector接口,配置如下(全类名)
a1.sources.r1.selector.type = org.example.MyChannelSelector

分享到:
评论

相关推荐

    flume1.7.0源码

    通过分析 Apache Flume 1.7.0 的源码,开发者可以深入了解其内部工作流程,优化性能,解决实际问题,并为构建自己的数据采集和传输解决方案打下坚实基础。同时,对于大数据领域的研究者,源码也是深入研究数据处理...

    大数据Ambari之flume集成编译好的源码包

    **大数据Ambari之flume集成编译好的源码包** Apache Ambari 是一个用于管理和监控Hadoop集群的开源工具,它提供了直观的Web界面和RESTful API,使得安装、配置、管理Hadoop生态系统变得更加简单。Flume是Apache的一...

    flume log4f示例源码

    Flume 是 Apache Hadoop 生态系统中的一个分布式、可靠且可用于有效收集、聚合和移动大量日志数据的...这样,你可以轻松地管理和分析大量的日志数据,尤其是在分布式环境中,Flume 提供了一种高效且可扩展的解决方案。

    Flume的channel和sink.docx

    - **多路复用/路由器Selector**:基于事件头属性进行路由选择,例如根据"class"属性将事件发送到不同的Channel。 - **Timestamp Interceptor**:添加或修改事件的时间戳。 - **Search And Replace Interceptor**:...

    基于Flume的分布式日志采集分析系统设计与实现.pdf

    基于Flume的分布式日志采集分析系统设计与实现 Flume是一种分布式日志采集系统,可以实时地采集和处理大量日志数据。该系统基于Flume、Elasticsearch和Kibana等技术手段,能够对海量日志数据进行实时采集、处理和...

    flume-jdbc-channel-1.5.0.1.zip

    【标题】"flume-jdbc-channel-1.5.0.1.zip" 涉及的是Apache Flume的一个特定版本,即1.5.0.1,与JDBC Channel的集成。Apache Flume是一个分布式、可靠且可用于有效收集、聚合和移动大量日志数据的系统。JDBC Channel...

    使用Flume收集数据内含源码以及说明书可以自己运行复现.zip

    这份资料“使用Flume收集数据内含源码以及说明书可以自己运行复现.zip”包含了实现Flume数据采集的源码和详细的操作指南,非常适合初学者深入理解和实践。 首先,让我们来理解Flume的原理。Apache Flume是一个高...

    尚硅谷大数据技术之Flume

    尚硅谷大数据技术之Flume Flume 是 Cloudera 提供的一个高可用的、 高可靠的、分布式的海量日志采集、聚合和传输的系统。 Flume 基于流式架构,灵活简单。 1.1 Flume 定义 Flume 是一个高可用的、 高可靠的、...

    数据采集之Flume.xmind

    Apache Flume 是一个分布式、高可靠、高可用的用来收集、聚合、转移不同来源的大量日志数据到中央数据仓库的工具 ...另外还有很多可选的组件interceptor、channel selector、sink processor等后面会介绍)。

    Apache flume1.6_src

    - 扩展性和可插拔架构:分析 Flume 如何支持自定义 Source、Sink 和 Channel 的开发。 源代码阅读可以提供对 Flume 工作原理的深刻理解,这对于优化 Flume 配置、排查问题或开发自定义组件都非常有价值。如果你想要...

    电商数仓项目(八) Flume(2) 拦截器开发源代码

    Java 是 Flume 源码的主要编程语言,它提供了一套丰富的 API 和接口,使得我们可以方便地扩展 Flume 的功能。在拦截器开发中,我们通常需要实现 `org.apache.flume.interceptor.Interceptor` 接口。这个接口包含两个...

    flume-ng安装

    在 `conf/flume-conf.properties` 文件中配置 Channel: ``` agent.channels.memoryChannel.type = memory agent.channels.memoryChannel.keep-alive = 30 agent.channels.memoryChannel.capacity = 10000 agent....

    flume-ng-sql-source-1.5.2

    - **可靠性**:通过使用可配置的持久化Channels(如File Channel),Flume可以在网络故障或节点故障后恢复未完成的传输,确保数据完整性。 - **灵活性**:Flume支持多种数据源和目标,可以通过配置轻松地扩展和...

    apache-flume-1.8.0

    在使用前,你需要根据你的需求配置Flume的配置文件,例如`conf/flume.conf`,定义数据流的源(source)、通道(channel)和接收器(sink)。 在部署Flume时,有几点需要注意: 1. **配置**: 配置文件定义了数据流动...

    大数据技术之Flume.docx

    大数据技术之Flume Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统。它基于流式架构,灵活简单。 Flume定义 Flume是一个高可用的,高可靠的,分布式的海量日志采集、...

    flume开发相关工具包

    总结来说,"flume开发相关工具包"是一个全面的资源集合,包含Flume的安装程序和开发所需的库,旨在帮助开发者构建和集成高效的数据流处理系统,尤其适合大数据环境下的日志管理和分析。无论是初学者还是经验丰富的...

    Flume集群环境搭建,flume监控

    Flume是一个由Cloudera公司开发的分布式、可靠且可用的系统,用于有效地收集、...通过以上内容的深入学习和实践,可以掌握Flume集群环境搭建和监控的全方位技能,为处理大规模日志数据的实时传输和分析打下坚实的基础。

Global site tag (gtag.js) - Google Analytics