a1.sources.r1.interceptors.i1.regex = (\\d):(\\d):(\\d)

a1.sources.r1.interceptors.i1.serializers = s1 s2 s3

a1.sources.r1.interceptors.i1.serializers.s1.name = one

a1.sources.r1.interceptors.i1.serializers.s2.name = two

a1.sources.r1.interceptors.i1.serializers.s3.name = three

a1.sources.r1.interceptors.i1.regex = ^(?:\\n)?(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)

a1.sources.r1.interceptors.i1.serializers = s1

a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer

a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp

a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm

#...

log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender

log4j.appender.flume.Hostname = example.com

log4j.appender.flume.Port =

log4j.appender.flume.UnsafeMode = true

# configure a class's logger to output to the flume appender

log4j.logger.org.example.MyClass = DEBUG,flume

#...

#...

log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender

log4j.appender.flume.Hostname = example.com

log4j.appender.flume.Port =

log4j.appender.flume.AvroReflectionEnabled = true

log4j.appender.flume.AvroSchemaUrl = hdfs://namenode/path/to/schema.avsc

# configure a class's logger to output to the flume appender

log4j.logger.org.example.MyClass = DEBUG,flume

#...

#...

log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender

log4j.appender.out2.Hosts = localhost: localhost:

# configure a class's logger to output to the flume appender

log4j.logger.org.example.MyClass = DEBUG,flume

#...

#...

log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender

log4j.appender.out2.Hosts = localhost: localhost:

log4j.appender.out2.Selector = RANDOM

# configure a class's logger to output to the flume appender

log4j.logger.org.example.MyClass = DEBUG,flume

#...

#...

log4j.appender.out2 = org.apache.flume.clients.log4jappender.LoadBalancingLog4jAppender

log4j.appender.out2.Hosts = localhost: localhost: localhost:

log4j.appender.out2.Selector = ROUND_ROBIN

log4j.appender.out2.MaxBackoff = 

# configure a class's logger to output to the flume appender

log4j.logger.org.example.MyClass = DEBUG,flume

#...

export JAVA_OPTS=”-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port= 
-Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false”

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=ganglia -Dflume.monitoring.hosts=com.example:,com.example2:

{

"typeName1.componentName1" : {"metric1" : "metricValue1", "metric2" : "metricValue2"},

"typeName2.componentName2" : {"metric3" : "metricValue3", "metric4" : "metricValue4"}

}

{

"CHANNEL.fileChannel":{"EventPutSuccessCount":"",

                      "Type":"CHANNEL",

                      "StopTime":"",

                      "EventPutAttemptCount":"",

                      "ChannelSize":"",

                      "StartTime":"",

                      "EventTakeSuccessCount":"",

                      "ChannelCapacity":"",

                      "EventTakeAttemptCount":""},

"CHANNEL.memChannel":{"EventPutSuccessCount":"",

                   "Type":"CHANNEL",

                   "StopTime":"",

                   "EventPutAttemptCount":"",

                   "ChannelSize":"",

                   "StartTime":"",

                   "EventTakeSuccessCount":"",

                   "ChannelCapacity":"",

                   "EventTakeAttemptCount":""}

}

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=http -Dflume.monitoring.port=

$ bin/flume-ng agent --conf-file example.conf --name a1 -Dflume.monitoring.type=com.example.reporting.HTTPReporting -Dflume.monitoring.node=com.example:

public class SinkCounter extends MonitoredCounterGroup implements

    SinkCounterMBean {

  private static final String COUNTER_CONNECTION_CREATED =

    "sink.connection.creation.count";

  private static final String COUNTER_CONNECTION_CLOSED =

    "sink.connection.closed.count";

  private static final String COUNTER_CONNECTION_FAILED =

    "sink.connection.failed.count";

  private static final String COUNTER_BATCH_EMPTY =

    "sink.batch.empty";

  private static final String COUNTER_BATCH_UNDERFLOW =

      "sink.batch.underflow";

  private static final String COUNTER_BATCH_COMPLETE =

    "sink.batch.complete";

  private static final String COUNTER_EVENT_DRAIN_ATTEMPT =

    "sink.event.drain.attempt";

  private static final String COUNTER_EVENT_DRAIN_SUCCESS =

    "sink.event.drain.sucess";

  private static final String[] ATTRIBUTES = {

    COUNTER_CONNECTION_CREATED, COUNTER_CONNECTION_CLOSED,

    COUNTER_CONNECTION_FAILED, COUNTER_BATCH_EMPTY,

    COUNTER_BATCH_UNDERFLOW, COUNTER_BATCH_COMPLETE,

    COUNTER_EVENT_DRAIN_ATTEMPT, COUNTER_EVENT_DRAIN_SUCCESS

  };

  public SinkCounter(String name) {

    super(MonitoredCounterGroup.Type.SINK, name, ATTRIBUTES);

  }

  @Override

  public long getConnectionCreatedCount() {

    return get(COUNTER_CONNECTION_CREATED);

  }

  public long incrementConnectionCreatedCount() {

    return increment(COUNTER_CONNECTION_CREATED);

  }

}

$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir

Event Validator Tool

事件验证器工具可用于以特定于应用程序的方式验证文件通道事件。该工具对每个事件应用用户提供程序验证登录，并删除不符合逻辑的事件。

工具可以运行如下:

$bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir -e org.apache.flume.MyEventValidator -DmaxSize 

其中datadir是要验证的数据目录的逗号分隔列表。

以下是可用的选项

事件验证器实现必须实现EventValidator接口。建议不要从实现中抛出任何异常，因为它们被视为无效事件。其他参数可以通过-D选项传递给EventValitor实现。

让我们看一个简单的基于大小的事件验证器示例，它将拒绝大于指定的最大大小的事件。

public static class MyEventValidator implements EventValidator {

  private int value = 0;

  private MyEventValidator(int val) {

    value = val;

  }

  @Override

  public boolean validateEvent(Event event) {

    return event.getBody() <= value;

  }

  public static class Builder implements EventValidator.Builder {

    private int sizeValidator = 0;

    @Override

    public EventValidator build() {

      return new DummyEventVerifier(sizeValidator);

    }

    @Override

    public void configure(Context context) {

      binaryValidator = context.getInteger("maxSize");

    }

  }

}

Topology Design Considerations

Flume非常灵活，允许大量可能的部署场景。如果您计划在大型生产部署中使用Flume，那么明智的做法是花一些时间考虑如何用Flume拓扑来表示问题。本节将介绍一些注意事项。

Is Flume a good fit for your problem?

如果您需要将文本日志数据导入Hadoop/HDFS中，那么Flume正好适合您的问题，完全停止。对于其他用例，这里有一些指导方针:

Flume的设计目的是在相对稳定、潜在复杂的拓扑结构上传输和摄取常规生成的事件数据。“事件数据”的概念定义非常广泛。对Flume来说，事件只是一个普通的字节blob。对于事件的大小有一些限制—例如，它不能大于您可以存储在内存或单个机器上的磁盘上的内容—但是在实践中，flume事件可以是从文本日志条目到图像文件的所有内容。事件的关键属性是以连续的流方式生成的。如果您的数据不是定期生成的(例如，您正在尝试将单个批量数据加载到Hadoop集群中)，那么Flume仍然可以工作，但是对于您的情况来说，这可能有些过头了。Flume喜欢相对稳定的拓扑结构。您的拓扑不需要是不可变的，因为Flume可以在不丢失数据的情况下处理拓扑中的更改，还可以容忍由于故障转移或供应而定期进行重新配置。如果您每天都尝试更改拓扑，那么它可能不会很好地工作，因为重新配置需要一些思考和开销。

Flow reliability in Flume

Flume流量的可靠性取决于几个因素。通过调整这些因素，您可以通过Flume实现广泛的可靠性选项。

你使用什么类型的频道?Flume既有持久通道(将数据持久化到磁盘上的通道)，也有非持久通道(如果机器发生故障将丢失数据的通道)。持久通道使用基于磁盘的存储，存储在此类通道中的数据将在机器重启或与磁盘无关的故障之间持续存在。

是否为工作负载提供了足够的通道。Flume中的通道在不同的跃点上充当缓冲器。这些缓冲器的容量是固定的，一旦容量满了，就会对流中较早的点产生反压力。如果这种压力传播到流的源头，水槽将不可用，可能会丢失数据。

是否使用冗余拓扑。Flume让您可以跨冗余拓扑复制流。这可以提供一个非常容易的容错源，克服磁盘或机器故障。

考虑Flume拓扑中的可靠性的最佳方法是考虑各种故障场景及其结果。如果磁盘出现故障怎么办?如果机器故障了怎么办?如果你的终端接收器(如HDFS)下降一段时间，你有背压，会发生什么?可能的设计空间很大，但是您需要问的基本问题却很少。

Flume topology design

设计Flume拓扑的第一步是枚举数据的所有源和目标(终端接收器)。这些将定义拓扑的边缘点。接下来要考虑的是是否引入中间聚合层或事件路由。如果您正在从大量源中收集数据，为了简化在终端接收器上的摄取，聚合这些数据是很有帮助的。聚合层还可以充当缓冲区，消除源的突发性或汇聚处的不可用性。如果您在不同位置之间路由数据，您可能还希望在不同的点上分割流:这将创建子拓扑，这些拓扑本身可能包含聚合点。

Sizing a Flume deployment

一旦您了解了拓扑的外观，下一个问题就是需要多少硬件和网络容量。首先要量化生成的数据量。这并不总是一项简单的任务!大多数数据流都是突发性的(例如，由于昼夜模式)，并且可能无法预测。一个好的起点是考虑拓扑的每一层的最大吞吐量，包括每秒的事件数和每秒的字节数。一旦您知道了给定层所需的吞吐量，就可以计算出该层需要多少节点的下限。为了确定可达到的吞吐量，最好在硬件上使用合成的或取样的事件数据对Flume进行试验。一般来说，基于磁盘的通道应该得到10 MB/s，基于内存的通道应该得到100 MB/s或更多。但是，根据硬件和操作环境的不同，性能差别很大。

调整聚合吞吐量的大小可以为每一层所需的节点数量提供一个下限。增加节点的原因有很多，比如增加冗余和更好地吸收负载中的突发事件。

Troubleshooting

Handling agent failures

如果Flume代理宕机，则该代理上承载的所有流都将中止。一旦代理重新启动，则流将恢复。使用文件流通道或其他稳定的通道将恢复处理事件离开。如果代理不能重新启动在相同的硬件上,然后有一个选项将数据库迁移到另一个硬件和设置一个新的水槽代理,可以保存在数据库恢复处理事件。可以利用数据库HA futures将Flume代理移动到另一个主机。

Compatibility

HDFS

目前Flume支持HDFS 0.20.2和0.23

完结

感谢有道翻译，主要是他的功能，我就是个搬砖的