Modified: websites/staging/flume/trunk/content/_sources/FlumeDeveloperGuide.txt
==============================================================================
--- websites/staging/flume/trunk/content/_sources/FlumeDeveloperGuide.txt
(original)
+++ websites/staging/flume/trunk/content/_sources/FlumeDeveloperGuide.txt Mon
Jun 1 19:50:47 2015
@@ -15,7 +15,7 @@
======================================
-Flume 1.5.2 Developer Guide
+Flume 1.6.0 Developer Guide
======================================
Introduction
@@ -277,6 +277,116 @@ properties:
request-timeout = 20000 # Must be >=1000 (default: 20000)
+Secure RPC client - Thrift
+''''''''''''''''''''''''''
+
+As of Flume 1.6.0, Thrift source and sink supports kerberos based
authentication.
+The client needs to use the getThriftInstance method of
``SecureRpcClientFactory``
+to get hold of a ``SecureThriftRpcClient``. ``SecureThriftRpcClient`` extends
+``ThriftRpcClient`` which implements the ``RpcClient`` interface. The kerberos
+authentication module resides in flume-ng-auth module which is
+required in classpath, when using the ``SecureRpcClientFactory``. Both the
client
+principal and the client keytab should be passed in as parameters through the
+properties and they reflect the credentials of the client to authenticate
+against the kerberos KDC. In addition, the server principal of the destination
+Thrift source to which this client is connecting to, should also be provided.
+The following example shows how to use the ``SecureRpcClientFactory``
+within a user's data-generating application:
+
+.. code-block:: java
+
+ import org.apache.flume.Event;
+ import org.apache.flume.EventDeliveryException;
+ import org.apache.flume.event.EventBuilder;
+ import org.apache.flume.api.SecureRpcClientFactory;
+ import org.apache.flume.api.RpcClientConfigurationConstants;
+ import org.apache.flume.api.RpcClient;
+ import java.nio.charset.Charset;
+ import java.util.Properties;
+
+ public class MyApp {
+ public static void main(String[] args) {
+ MySecureRpcClientFacade client = new MySecureRpcClientFacade();
+ // Initialize client with the remote Flume agent's host, port
+ Properties props = new Properties();
+ props.setProperty(RpcClientConfigurationConstants.CONFIG_CLIENT_TYPE,
"thrift");
+ props.setProperty("hosts", "h1");
+ props.setProperty("hosts.h1", "client.example.org"+":"+
String.valueOf(41414));
+
+ // Initialize client with the kerberos authentication related properties
+ props.setProperty("kerberos", "true");
+ props.setProperty("client-principal",
"flumeclient/[email protected]");
+ props.setProperty("client-keytab", "/tmp/flumeclient.keytab");
+ props.setProperty("server-principal",
"flume/[email protected]");
+ client.init(props);
+
+ // 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 MySecureRpcClientFacade {
+ private RpcClient client;
+ private Properties properties;
+
+ public void init(Properties properties) {
+ // Setup the RPC connection
+ this.properties = properties;
+ // Create the ThriftSecureRpcClient instance by using
SecureRpcClientFactory
+ this.client = SecureRpcClientFactory.getThriftInstance(properties);
+ }
+
+ 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 = SecureRpcClientFactory.getThriftInstance(properties);
+ }
+ }
+
+ public void cleanUp() {
+ // Close the RPC connection
+ client.close();
+ }
+ }
+
+The remote ``ThriftSource`` should be started in kerberos mode.
+Below is an example Flume agent configuration that's waiting for a connection
+from MyApp:
+
+.. code-block:: properties
+
+ a1.channels = c1
+ a1.sources = r1
+ a1.sinks = k1
+
+ a1.channels.c1.type = memory
+
+ a1.sources.r1.channels = c1
+ a1.sources.r1.type = thrift
+ a1.sources.r1.bind = 0.0.0.0
+ a1.sources.r1.port = 41414
+ a1.sources.r1.kerberos = true
+ a1.sources.r1.agent-principal = flume/[email protected]
+ a1.sources.r1.agent-keytab = /tmp/flume.keytab
+
+
+ a1.sinks.k1.channel = c1
+ a1.sinks.k1.type = logger
+
Failover Client
'''''''''''''''
@@ -450,7 +560,7 @@ sources, sinks, and channels are allowed
is a special embedded source and events should be send to the source
via the put, putAll methods on the EmbeddedAgent object. Only File Channel
and Memory Channel are allowed as channels while Avro Sink is the only
-supported sink.
+supported sink. Interceptors are also supported by the embedded agent.
Note: The embedded agent has a dependency on hadoop-core.jar.
@@ -459,18 +569,29 @@ full Agent. The following is an exhausti
Required properties are in **bold**.
-==================== ================
==============================================
-Property Name Default Description
-==================== ================
==============================================
-source.type embedded The only available source is the
embedded source.
-**channel.type** -- Either ``memory`` or ``file`` which
correspond to MemoryChannel and FileChannel respectively.
-channel.* -- Configuration options for the channel
type requested, see MemoryChannel or FileChannel user guide for an exhaustive
list.
-**sinks** -- List of sink names
-**sink.type** -- Property name must match a name in the
list of sinks. Value must be ``avro``
-sink.* -- Configuration options for the sink.
See AvroSink user guide for an exhaustive list, however note AvroSink requires
at least hostname and port.
-**processor.type** -- Either ``failover`` or
``load_balance`` which correspond to FailoverSinksProcessor and
LoadBalancingSinkProcessor respectively.
-processor.* -- Configuration options for the sink
processor selected. See FailoverSinksProcessor and LoadBalancingSinkProcessor
user guide for an exhaustive list.
-==================== ================
==============================================
+===================== ================
======================================================================
+Property Name Default Description
+===================== ================
======================================================================
+source.type embedded The only available source is the
embedded source.
+**channel.type** -- Either ``memory`` or ``file`` which
correspond
+ to MemoryChannel and FileChannel
respectively.
+channel.* -- Configuration options for the channel
type requested,
+ see MemoryChannel or FileChannel user
guide for an exhaustive list.
+**sinks** -- List of sink names
+**sink.type** -- Property name must match a name in
the list of sinks.
+ Value must be ``avro``
+sink.* -- Configuration options for the sink.
+ See AvroSink user guide for an
exhaustive list,
+ however note AvroSink requires at
least hostname and port.
+**processor.type** -- Either ``failover`` or
``load_balance`` which correspond
+ to FailoverSinksProcessor and
LoadBalancingSinkProcessor respectively.
+processor.* -- Configuration options for the sink
processor selected.
+ See FailoverSinksProcessor and
LoadBalancingSinkProcessor
+ user guide for an exhaustive list.
+source.interceptors -- Space-separated list of interceptors
+source.interceptors.* -- Configuration options for individual
interceptors
+ specified in the source.interceptors
property
+===================== ================
======================================================================
Below is an example of how to use the agent:
@@ -487,6 +608,10 @@ Below is an example of how to use the ag
properties.put("sink2.hostname", "collector2.apache.org");
properties.put("sink2.port", "5565");
properties.put("processor.type", "load_balance");
+ properties.put("source.interceptors", "i1");
+ properties.put("source.interceptors.i1.type", "static");
+ properties.put("source.interceptors.i1.key", "key1");
+ properties.put("source.interceptors.i1.value", "value1");
EmbeddedAgent agent = new EmbeddedAgent("myagent");
Modified: websites/staging/flume/trunk/content/_sources/FlumeUserGuide.txt
==============================================================================
--- websites/staging/flume/trunk/content/_sources/FlumeUserGuide.txt (original)
+++ websites/staging/flume/trunk/content/_sources/FlumeUserGuide.txt Mon Jun 1
19:50:47 2015
@@ -15,7 +15,7 @@
======================================
-Flume 1.5.2 User Guide
+Flume 1.6.0 User Guide
======================================
Introduction
@@ -234,6 +234,31 @@ The original Flume terminal will output
Congratulations - you've successfully configured and deployed a Flume agent!
Subsequent sections cover agent configuration in much more detail.
+
+Zookeeper based Configuration
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Flume supports Agent configurations via Zookeeper. *This is an experimental
feature.* The configuration file needs to be uploaded
+in the Zookeeper, under a configurable prefix. The configuration file is
stored in Zookeeper Node data.
+Following is how the Zookeeper Node tree would look like for agents a1 and a2
+
+.. code-block:: properties
+
+ - /flume
+ |- /a1 [Agent config file]
+ |- /a2 [Agent config file]
+
+Once the configuration file is uploaded, start the agent with following options
+
+ $ bin/flume-ng agent --conf conf -z zkhost:2181,zkhost1:2181 -p /flume
--name a1 -Dflume.root.logger=INFO,console
+
+================== ================
=========================================================================
+Argument Name Default Description
+================== ================
=========================================================================
+**z** -- Zookeeper connection string. Comma
separated list of hostname:port
+**p** /flume Base Path in Zookeeper to store Agent
configurations
+================== ================
=========================================================================
+
Installing third-party plugins
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -707,7 +732,7 @@ keystore-password -- T
keystore-type JKS The type of the Java keystore. This can
be "JKS" or "PKCS12".
exclude-protocols SSLv3 Space-separated list of SSL/TLS
protocols to exclude. SSLv3 will always be excluded in addition to the
protocols specified.
ipFilter false Set this to true to enable ipFiltering
for netty
-ipFilter.rules -- Define N netty ipFilter pattern rules
with this config.
+ipFilterRules -- Define N netty ipFilter pattern rules
with this config.
================== ================
===================================================
Example for agent named a1:
@@ -721,15 +746,15 @@ Example for agent named a1:
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141
-Example of ipFilter.rules
+Example of ipFilterRules
-ipFilter.rules defines N netty ipFilters separated by a comma a pattern rule
must be in this format.
+ipFilterRules defines N netty ipFilters separated by a comma a pattern rule
must be in this format.
<'allow' or deny>:<'ip' or 'name' for computer name>:<pattern>
or
allow/deny:ip/name:pattern
-example: ipFilter.rules=allow:ip:127.*,allow:name:localhost,deny:ip:*
+example: ipFilterRules=allow:ip:127.*,allow:name:localhost,deny:ip:*
Note that the first rule to match will apply as the example below shows from a
client on the localhost
@@ -742,6 +767,9 @@ Thrift Source
Listens on Thrift port and receives events from external Thrift client streams.
When paired with the built-in ThriftSink on another (previous hop) Flume agent,
it can create tiered collection topologies.
+Thrift source can be configured to start in secure mode by enabling kerberos
authentication.
+agent-principal and agent-keytab are the properties used by the
+Thrift source to authenticate to the kerberos KDC.
Required properties are in **bold**.
================== ===========
===================================================
@@ -756,6 +784,14 @@ selector.type
selector.*
interceptors -- Space separated list of interceptors
interceptors.*
+ssl false Set this to true to enable SSL encryption.
You must also specify a "keystore" and a "keystore-password".
+keystore -- This is the path to a Java keystore file.
Required for SSL.
+keystore-password -- The password for the Java keystore. Required
for SSL.
+keystore-type JKS The type of the Java keystore. This can be
"JKS" or "PKCS12".
+exclude-protocols SSLv3 Space-separated list of SSL/TLS protocols to
exclude. SSLv3 will always be excluded in addition to the protocols specified.
+kerberos false Set to true to enable kerberos
authentication. In kerberos mode, agent-principal and agent-keytab are
required for successful authentication. The Thrift source in secure mode, will
accept connections only from Thrift clients that have kerberos enabled and are
successfully authenticated to the kerberos KDC.
+agent-principal -- The kerberos principal used by the Thrift
Source to authenticate to the kerberos KDC.
+agent-keytab â- The keytab location used by the Thrift
Source in combination with the agent-principal to authenticate to the kerberos
KDC.
================== ===========
===================================================
Example for agent named a1:
@@ -793,6 +829,7 @@ restartThrottle 10000 Amount of
restart false Whether the executed cmd should be restarted if
it dies
logStdErr false Whether the command's stderr should be logged
batchSize 20 The max number of lines to read and send to the
channel at a time
+batchTimeout 3000 Amount of time (in milliseconds) to wait, if the
buffer size was not reached, before data is pushed downstream
selector.type replicating replicating or multiplexing
selector.* Depends on the selector.type value
interceptors -- Space-separated list of interceptors
@@ -841,9 +878,9 @@ invoked directly. Common values for 'sh
.. code-block:: properties
- agent_foo.sources.tailsource-1.type = exec
- agent_foo.sources.tailsource-1.shell = /bin/bash -c
- agent_foo.sources.tailsource-1.command = for i in /path/*.txt; do cat $i;
done
+ a1.sources.tailsource-1.type = exec
+ a1.sources.tailsource-1.shell = /bin/bash -c
+ a1.sources.tailsource-1.command = for i in /path/*.txt; do cat $i; done
JMS Source
~~~~~~~~~~~
@@ -986,54 +1023,13 @@ Example for an agent named agent-1:
.. code-block:: properties
- agent-1.channels = ch-1
- agent-1.sources = src-1
+ a1.channels = ch-1
+ a1.sources = src-1
- agent-1.sources.src-1.type = spooldir
- agent-1.sources.src-1.channels = ch-1
- agent-1.sources.src-1.spoolDir = /var/log/apache/flumeSpool
- agent-1.sources.src-1.fileHeader = true
-
-Twitter 1% firehose Source (experimental)
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-.. warning::
- This source is hightly experimental and may change between minor versions of
Flume.
- Use at your own risk.
-
-Experimental source that connects via Streaming API to the 1% sample twitter
-firehose, continously downloads tweets, converts them to Avro format and
-sends Avro events to a downstream Flume sink. Requires the consumer and
-access tokens and secrets of a Twitter developer account.
-Required properties are in **bold**.
-
-====================== ===========
===================================================
-Property Name Default Description
-====================== ===========
===================================================
-**channels** --
-**type** -- The component type name, needs to be
``org.apache.flume.source.twitter.TwitterSource``
-**consumerKey** -- OAuth consumer key
-**consumerSecret** -- OAuth consumer secret
-**accessToken** -- OAuth access token
-**accessTokenSecret** -- OAuth toekn secret
-maxBatchSize 1000 Maximum number of twitter messages to put
in a single batch
-maxBatchDurationMillis 1000 Maximum number of milliseconds to wait
before closing a batch
-====================== ===========
===================================================
-
-Example for agent named a1:
-
-.. code-block:: properties
-
- a1.sources = r1
- a1.channels = c1
- a1.sources.r1.type = org.apache.flume.source.twitter.TwitterSource
- a1.sources.r1.channels = c1
- a1.sources.r1.consumerKey = YOUR_TWITTER_CONSUMER_KEY
- a1.sources.r1.consumerSecret = YOUR_TWITTER_CONSUMER_SECRET
- a1.sources.r1.accessToken = YOUR_TWITTER_ACCESS_TOKEN
- a1.sources.r1.accessTokenSecret = YOUR_TWITTER_ACCESS_TOKEN_SECRET
- a1.sources.r1.maxBatchSize = 10
- a1.sources.r1.maxBatchDurationMillis = 200
+ a1.sources.src-1.type = spooldir
+ a1.sources.src-1.channels = ch-1
+ a1.sources.src-1.spoolDir = /var/log/apache/flumeSpool
+ a1.sources.src-1.fileHeader = true
Event Deserializers
'''''''''''''''''''
@@ -1094,6 +1090,95 @@ Property Name Default
deserializer.maxBlobLength 100000000 The maximum number of bytes to
read and buffer for a given request
========================== ==================
=======================================================================
+Twitter 1% firehose Source (experimental)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. warning::
+ This source is hightly experimental and may change between minor versions of
Flume.
+ Use at your own risk.
+
+Experimental source that connects via Streaming API to the 1% sample twitter
+firehose, continously downloads tweets, converts them to Avro format and
+sends Avro events to a downstream Flume sink. Requires the consumer and
+access tokens and secrets of a Twitter developer account.
+Required properties are in **bold**.
+
+====================== ===========
===================================================
+Property Name Default Description
+====================== ===========
===================================================
+**channels** --
+**type** -- The component type name, needs to be
``org.apache.flume.source.twitter.TwitterSource``
+**consumerKey** -- OAuth consumer key
+**consumerSecret** -- OAuth consumer secret
+**accessToken** -- OAuth access token
+**accessTokenSecret** -- OAuth toekn secret
+maxBatchSize 1000 Maximum number of twitter messages to put
in a single batch
+maxBatchDurationMillis 1000 Maximum number of milliseconds to wait
before closing a batch
+====================== ===========
===================================================
+
+Example for agent named a1:
+
+.. code-block:: properties
+
+ a1.sources = r1
+ a1.channels = c1
+ a1.sources.r1.type = org.apache.flume.source.twitter.TwitterSource
+ a1.sources.r1.channels = c1
+ a1.sources.r1.consumerKey = YOUR_TWITTER_CONSUMER_KEY
+ a1.sources.r1.consumerSecret = YOUR_TWITTER_CONSUMER_SECRET
+ a1.sources.r1.accessToken = YOUR_TWITTER_ACCESS_TOKEN
+ a1.sources.r1.accessTokenSecret = YOUR_TWITTER_ACCESS_TOKEN_SECRET
+ a1.sources.r1.maxBatchSize = 10
+ a1.sources.r1.maxBatchDurationMillis = 200
+
+Kafka Source
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Kafka Source is an Apache Kafka consumer that reads messages from a Kafka
topic.
+If you have multiple Kafka sources running, you can configure them with the
same Consumer Group
+so each will read a unique set of partitions for the topic.
+
+
+
+=============================== ===========
===================================================
+Property Name Default Description
+=============================== ===========
===================================================
+**channels** --
+**type** -- The component type name, needs
to be ``org.apache.flume.source.kafka,KafkaSource``
+**zookeeperConnect** -- URI of ZooKeeper used by Kafka
cluster
+**groupId** flume Unique identified of consumer
group. Setting the same id in multiple sources or agents
+ indicates that they are part of
the same consumer group
+**topic** -- Kafka topic we'll read messages
from. At the time, this is a single topic only.
+batchSize 1000 Maximum number of messages
written to Channel in one batch
+batchDurationMillis 1000 Maximum time (in ms) before a
batch will be written to Channel
+ The batch will be written
whenever the first of size and time will be reached.
+Other Kafka Consumer Properties -- These properties are used to
configure the Kafka Consumer. Any producer property supported
+ by Kafka can be used. The only
requirement is to prepend the property name with the prefix ``kafka.``.
+ For example:
kafka.consumer.timeout.ms
+ Check `Kafka documentation
<https://kafka.apache.org/08/configuration.html#consumerconfigs>` for details
+=============================== ===========
===================================================
+
+.. note:: The Kafka Source overrides two Kafka consumer parameters:
+ auto.commit.enable is set to "false" by the source and we commit
every batch. For improved performance
+ this can be set to "true", however, this can lead to loss of data
+ consumer.timeout.ms is set to 10ms, so when we check Kafka for new
data we wait at most 10ms for the data to arrive
+ setting this to a higher value can reduce CPU utilization (we'll
poll Kafka in less of a tight loop), but also means
+ higher latency in writing batches to channel (since we'll wait
longer for data to arrive).
+
+
+Example for agent named tier1:
+
+.. code-block:: properties
+
+ tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
+ tier1.sources.source1.channels = channel1
+ tier1.sources.source1.zookeeperConnect = localhost:2181
+ tier1.sources.source1.topic = test1
+ tier1.sources.source1.groupId = flume
+ tier1.sources.source1.kafka.consumer.timeout.ms = 100
+
+
+
NetCat Source
~~~~~~~~~~~~~
@@ -1182,8 +1267,13 @@ Property Name Default Descriptio
**host** -- Host name or IP address to bind to
**port** -- Port # to bind to
eventSize 2500 Maximum size of a single event line, in bytes
-keepFields false Setting this to true will preserve the Priority,
+keepFields none Setting this to 'all' will preserve the Priority,
Timestamp and Hostname in the body of the event.
+ A spaced separated list of fields to include
+ is allowed as well. Currently, the following
+ fields can be included: priority, version,
+ timestamp, hostname. The values 'true' and
'false'
+ have been deprecated in favor of 'all' and
'none'.
selector.type replicating or multiplexing
selector.* replicating Depends on the selector.type value
interceptors -- Space-separated list of interceptors
@@ -1220,8 +1310,13 @@ Property Name Default
**host** -- Host name or IP address to bind to.
**ports** -- Space-separated list (one or more) of
ports to bind to.
eventSize 2500 Maximum size of a single event line,
in bytes.
-keepFields false Setting this to true will preserve the
+keepFields none Setting this to 'all' will preserve the
Priority, Timestamp and Hostname in
the body of the event.
+ A spaced separated list of fields to
include
+ is allowed as well. Currently, the
following
+ fields can be included: priority,
version,
+ timestamp, hostname. The values 'true'
and 'false'
+ have been deprecated in favor of 'all'
and 'none'.
portHeader -- If specified, the port number will be
stored in the header of each event using the header name specified here. This
allows for interceptors and channel selectors to customize routing logic based
on the incoming port.
charset.default UTF-8 Default character set used while
parsing syslog events into strings.
charset.port.<port> -- Character set is configurable on a
per-port basis.
@@ -1374,6 +1469,37 @@ Property Name Default
handler.maxBlobLength 100000000 The maximum number of bytes to read
and buffer for a given request
===================== ==================
============================================================================
+Stress Source
+~~~~~~~~~~~~~
+
+StressSource is an internal load-generating source implementation which is
very useful for
+stress tests. It allows User to configure the size of Event payload, with
empty headers.
+User can configure total number of events to be sent as well maximum number of
Successful
+Event to be delivered.
+
+Required properties are in **bold**.
+
+=================== ===========
===================================================
+Property Name Default Description
+=================== ===========
===================================================
+**type** -- The component type name, needs to be
``org.apache.flume.source.StressSource``
+size 500 Payload size of each Event. Unit:**byte**
+maxTotalEvents -1 Maximum number of Events to be sent
+maxSuccessfulEvents -1 Maximum number of Events successfully sent
+batchSize 1 Number of Events to be sent in one batch
+=================== ===========
===================================================
+
+Example for agent named **a1**:
+
+.. code-block:: properties
+
+ a1.sources = stresssource-1
+ a1.channels = memoryChannel-1
+ a1.sources.stresssource-1.type = org.apache.flume.source.StressSource
+ a1.sources.stresssource-1.size = 10240
+ a1.sources.stresssource-1.maxTotalEvents = 1000000
+ a1.sources.stresssource-1.channels = memoryChannel-1
+
Legacy Sources
~~~~~~~~~~~~~~
@@ -1485,15 +1611,16 @@ Flume should use ScribeSource based on T
For deployment of Scribe please follow the guide from Facebook.
Required properties are in **bold**.
-============== =========== ==============================================
-Property Name Default Description
-============== =========== ==============================================
-**type** -- The component type name, needs to be
``org.apache.flume.source.scribe.ScribeSource``
-port 1499 Port that Scribe should be connected
-workerThreads 5 Handing threads number in Thrift
+==================== ===========
==============================================
+Property Name Default Description
+==================== ===========
==============================================
+**type** -- The component type name, needs to be
``org.apache.flume.source.scribe.ScribeSource``
+port 1499 Port that Scribe should be connected
+maxReadBufferBytes 16384000 Thrift Default FrameBuffer Size
+workerThreads 5 Handing threads number in Thrift
selector.type
selector.*
-============== =========== ==============================================
+==================== ===========
==============================================
Example for agent named a1:
@@ -1537,12 +1664,14 @@ Alias Description
%B locale's long month name (January, February, ...)
%c locale's date and time (Thu Mar 3 23:05:25 2005)
%d day of month (01)
+%e day of month without padding (1)
%D date; same as %m/%d/%y
%H hour (00..23)
%I hour (01..12)
%j day of year (001..366)
%k hour ( 0..23)
%m month (01..12)
+%n month without padding (1..12)
%M minute (00..59)
%p locale's equivalent of am or pm
%s seconds since 1970-01-01 00:00:00 UTC
@@ -1599,10 +1728,10 @@ hdfs.roundValue 1 Ro
hdfs.roundUnit second The unit of the round down value -
``second``, ``minute`` or ``hour``.
hdfs.timeZone Local Time Name of the timezone that should be used
for resolving the directory path, e.g. America/Los_Angeles.
hdfs.useLocalTimeStamp false Use the local time (instead of the
timestamp from the event header) while replacing the escape sequences.
-hdfs.closeTries 0 Number of times the sink must try to
close a file. If set to 1, this sink will not re-try a failed close
+hdfs.closeTries 0 Number of times the sink must try
renaming a file, after initiating a close attempt. If set to 1, this sink will
not re-try a failed rename
(due to, for example, NameNode or
DataNode failure), and may leave the file in an open state with a .tmp
extension.
- If set to 0, the sink will try to close
the file until the file is eventually closed
- (there is no limit on the number of
times it would try).
+ If set to 0, the sink will try to rename
the file until the file is eventually renamed (there is no limit on the number
of times it would try).
+ The file may still remain open if the
close call fails but the data will be intact and in this case, the file will be
closed only after a Flume restart.
hdfs.retryInterval 180 Time in seconds between consecutive
attempts to close a file. Each close call costs multiple RPC round-trips to the
Namenode,
so setting this too low can cause a lot
of load on the name node. If set to 0 or less, the sink will not
attempt to close the file if the first
attempt fails, and may leave the file open or with a ".tmp" extension.
@@ -1630,6 +1759,149 @@ The above configuration will round down
timestamp 11:54:34 AM, June 12, 2012 will cause the hdfs path to become
``/flume/events/2012-06-12/1150/00``.
+Hive Sink
+~~~~~~~~~
+
+This sink streams events containing delimited text or JSON data directly into
a Hive table or partition.
+Events are written using Hive transactions. As soon as a set of events are
committed to Hive, they become
+immediately visible to Hive queries. Partitions to which flume will stream to
can either be pre-created
+or, optionally, Flume can create them if they are missing. Fields from
incoming event data are mapped to
+corresponding columns in the Hive table. **This sink is provided as a preview
feature and not recommended
+for use in production.**
+
+====================== ============
======================================================================
+Name Default Description
+====================== ============
======================================================================
+**channel** --
+**type** -- The component type name, needs to be
``hive``
+**hive.metastore** -- Hive metastore URI (eg
thrift://a.b.com:9083 )
+**hive.database** -- Hive database name
+**hive.table** -- Hive table name
+hive.partition -- Comma separate list of partition
values identifying the partition to write to. May contain escape
+ sequences. E.g: If the table is
partitioned by (continent: string, country :string, time : string)
+ then 'Asia,India,2014-02-26-01-21'
will indicate continent=Asia,country=India,time=2014-02-26-01-21
+hive.txnsPerBatchAsk 100 Hive grants a *batch of transactions*
instead of single transactions to streaming clients like Flume.
+ This setting configures the number of
desired transactions per Transaction Batch. Data from all
+ transactions in a single batch end up
in a single file. Flume will write a maximum of batchSize events
+ in each transaction in the batch. This
setting in conjunction with batchSize provides control over the
+ size of each file. Note that
eventually Hive will transparently compact these files into larger files.
+heartBeatInterval 240 (In seconds) Interval between
consecutive heartbeats sent to Hive to keep unused transactions from expiring.
+ Set this value to 0 to disable
heartbeats.
+autoCreatePartitions true Flume will automatically create the
necessary Hive partitions to stream to
+batchSize 15000 Max number of events written to Hive
in a single Hive transaction
+maxOpenConnections 500 Allow only this number of open
connections. If this number is exceeded, the least recently used connection is
closed.
+callTimeout 10000 (In milliseconds) Timeout for Hive &
HDFS I/O operations, such as openTxn, write, commit, abort.
+**serializer** Serializer is responsible for parsing
out field from the event and mapping them to columns in the hive table.
+ Choice of serializer depends upon the
format of the data in the event. Supported serializers: DELIMITED and JSON
+roundUnit minute The unit of the round down value -
``second``, ``minute`` or ``hour``.
+roundValue 1 Rounded down to the highest multiple
of this (in the unit configured using hive.roundUnit), less than current time
+timeZone Local Time Name of the timezone that should be
used for resolving the escape sequences in partition, e.g. America/Los_Angeles.
+useLocalTimeStamp false Use the local time (instead of the
timestamp from the event header) while replacing the escape sequences.
+====================== ============
======================================================================
+
+Following serializers are provided for Hive sink:
+
+**JSON**: Handles UTF8 encoded Json (strict syntax) events and requires no
configration. Object names
+in the JSON are mapped directly to columns with the same name in the Hive
table.
+Internally uses org.apache.hive.hcatalog.data.JsonSerDe but is independent of
the Serde of the Hive table.
+This serializer requires HCatalog to be installed.
+
+**DELIMITED**: Handles simple delimited textual events.
+Internally uses LazySimpleSerde but is independent of the Serde of the Hive
table.
+
+========================== ============
======================================================================
+Name Default Description
+========================== ============
======================================================================
+serializer.delimiter , (Type: string) The field delimiter
in the incoming data. To use special
+ characters, surround them with
double quotes like "\\t"
+**serializer.fieldnames** -- The mapping from input fields to
columns in hive table. Specified as a
+ comma separated list (no spaces)
of hive table columns names, identifying
+ the input fields in order of their
occurrence. To skip fields leave the
+ column name unspecified. Eg.
'time,,ip,message' indicates the 1st, 3rd
+ and 4th fields in input map to
time, ip and message columns in the hive table.
+serializer.serdeSeparator Ctrl-A (Type: character) Customizes the
separator used by underlying serde. There
+ can be a gain in efficiency if the
fields in serializer.fieldnames are in
+ same order as table columns, the
serializer.delimiter is same as the
+ serializer.serdeSeparator and
number of fields in serializer.fieldnames
+ is less than or equal to number of
table columns, as the fields in incoming
+ event body do not need to be
reordered to match order of table columns.
+ Use single quotes for special
characters like '\\t'.
+ Ensure input fields do not contain
this character. NOTE: If serializer.delimiter
+ is a single character, preferably
set this to the same character
+========================== ============
======================================================================
+
+
+The following are the escape sequences supported:
+
+========= =================================================
+Alias Description
+========= =================================================
+%{host} Substitute value of event header named "host". Arbitrary header
names are supported.
+%t Unix time in milliseconds
+%a locale's short weekday name (Mon, Tue, ...)
+%A locale's full weekday name (Monday, Tuesday, ...)
+%b locale's short month name (Jan, Feb, ...)
+%B locale's long month name (January, February, ...)
+%c locale's date and time (Thu Mar 3 23:05:25 2005)
+%d day of month (01)
+%D date; same as %m/%d/%y
+%H hour (00..23)
+%I hour (01..12)
+%j day of year (001..366)
+%k hour ( 0..23)
+%m month (01..12)
+%M minute (00..59)
+%p locale's equivalent of am or pm
+%s seconds since 1970-01-01 00:00:00 UTC
+%S second (00..59)
+%y last two digits of year (00..99)
+%Y year (2010)
+%z +hhmm numeric timezone (for example, -0400)
+========= =================================================
+
+
+.. note:: For all of the time related escape sequences, a header with the key
+ "timestamp" must exist among the headers of the event (unless
``useLocalTimeStamp`` is set to ``true``). One way to add
+ this automatically is to use the TimestampInterceptor.
+
+Example Hive table :
+
+.. code-block:: properties
+
+ create table weblogs ( id int , msg string )
+ partitioned by (continent string, country string, time string)
+ clustered by (id) into 5 buckets
+ stored as orc;
+
+Example for agent named a1:
+
+.. code-block:: properties
+
+ a1.channels = c1
+ a1.channels.c1.type = memory
+ a1.sinks = k1
+ a1.sinks.k1.type = hive
+ a1.sinks.k1.channel = c1
+ a1.sinks.k1.hive.metastore = thrift://127.0.0.1:9083
+ a1.sinks.k1.hive.database = logsdb
+ a1.sinks.k1.hive.table = weblogs
+ a1.sinks.k1.hive.partition = asia,%{country},%y-%m-%d-%H-%M
+ a1.sinks.k1.useLocalTimeStamp = false
+ a1.sinks.k1.round = true
+ a1.sinks.k1.roundValue = 10
+ a1.sinks.k1.roundUnit = minute
+ a1.sinks.k1.serializer = DELIMITED
+ a1.sinks.k1.serializer.delimiter = "\t"
+ a1.sinks.k1.serializer.serdeSeparator = '\t'
+ a1.sinks.k1.serializer.fieldnames =id,,msg
+
+
+The above configuration will round down the timestamp to the last 10th minute.
For example, an event with
+timestamp header set to 11:54:34 AM, June 12, 2012 and 'country' header set to
'india' will evaluate to the
+partition (continent='asia',country='india',time='2012-06-12-11-50'. The
serializer is configured to
+accept tab separated input containing three fields and to skip the second
field.
+
+
Logger Sink
~~~~~~~~~~~
@@ -1641,6 +1913,7 @@ Property Name Default Description
============== ======= ===========================================
**channel** --
**type** -- The component type name, needs to be ``logger``
+maxBytesToLog 16 Maximum number of bytes of the Event body to log
============== ======= ===========================================
Example for agent named a1:
@@ -1701,6 +1974,12 @@ This sink forms one half of Flume's tier
sent to this sink are turned into Thrift events and sent to the configured
hostname / port pair. The events are taken from the configured Channel in
batches of the configured batch size.
+
+Thrift sink can be configured to start in secure mode by enabling kerberos
authentication.
+To communicate with a Thrift source started in secure mode, the Thrift sink
should also
+operate in secure mode. client-principal and client-keytab are the properties
used by the
+Thrift sink to authenticate to the kerberos KDC. The server-principal
represents the
+principal of the Thrift source this sink is configured to connect to in secure
mode.
Required properties are in **bold**.
========================== =======
==============================================
@@ -1714,6 +1993,15 @@ batch-size 100 nu
connect-timeout 20000 Amount of time (ms) to allow for the
first (handshake) request.
request-timeout 20000 Amount of time (ms) to allow for
requests after the first.
connection-reset-interval none Amount of time (s) before the connection
to the next hop is reset. This will force the Thrift Sink to reconnect to the
next hop. This will allow the sink to connect to hosts behind a hardware
load-balancer when news hosts are added without having to restart the agent.
+ssl false Set to true to enable SSL for this
ThriftSink. When configuring SSL, you can optionally set a "truststore",
"truststore-password" and "truststore-type"
+truststore -- The path to a custom Java truststore
file. Flume uses the certificate authority information in this file to
determine whether the remote Thrift Source's SSL authentication credentials
should be trusted. If not specified, the default Java JSSE certificate
authority files (typically "jssecacerts" or "cacerts" in the Oracle JRE) will
be used.
+truststore-password -- The password for the specified
truststore.
+truststore-type JKS The type of the Java truststore. This
can be "JKS" or other supported Java truststore type.
+exclude-protocols SSLv3 Space-separated list of SSL/TLS
protocols to exclude
+kerberos false Set to true to enable kerberos
authentication. In kerberos mode, client-principal, client-keytab and
server-principal are required for successful authentication and communication
to a kerberos enabled Thrift Source.
+client-principal â- The kerberos principal used by the
Thrift Sink to authenticate to the kerberos KDC.
+client-keytab â- The keytab location used by the Thrift
Sink in combination with the client-principal to authenticate to the kerberos
KDC.
+server-principal -- The kerberos principal of the Thrift
Source to which the Thrift Sink is configured to connect to.
========================== =======
==============================================
Example for agent named a1:
@@ -2011,7 +2299,9 @@ Property Name Default
**type** --
The component type name, needs to be
``org.apache.flume.sink.elasticsearch.ElasticSearchSink``
**hostNames** --
Comma separated list of hostname:port, if the port is not present
the default port '9300' will be used
indexName flume
The name of the index which the date will be appended to. Example
'flume' -> 'flume-yyyy-MM-dd'
+
Arbitrary header substitution is supported, eg. %{header} replaces
with value of named event header
indexType logs
The type to index the document to, defaults to 'log'
+
Arbitrary header substitution is supported, eg. %{header} replaces
with value of named event header
clusterName elasticsearch
Name of the ElasticSearch cluster to connect to
batchSize 100
Number of events to be written per txn.
ttl --
TTL in days, when set will cause the expired documents to be
deleted automatically,
@@ -2024,6 +2314,10 @@ serializer org.apache.flume.sink.
serializer.* --
Properties to be passed to the serializer.
================
========================================================================
=======================================================================================================
+.. note:: Header substitution is a handy to use the value of an event header
to dynamically decide the indexName and indexType to use when storing the event.
+ Caution should be used in using this feature as the event submitter
now has control of the indexName and indexType.
+ Furthermore, if the elasticsearch REST client is used then the event
submitter has control of the URL path used.
+
Example for agent named a1:
.. code-block:: properties
@@ -2040,18 +2334,13 @@ Example for agent named a1:
a1.sinks.k1.serializer =
org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
a1.sinks.k1.channel = c1
-Kite Dataset Sink (experimental)
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-.. warning::
- This source is experimental and may change between minor versions of Flume.
- Use at your own risk.
+Kite Dataset Sink
+~~~~~~~~~~~~~~~~~
-Experimental sink that writes events to a `Kite Dataset
<http://kitesdk.org/docs/current/kite-data/guide.html>`_.
+Experimental sink that writes events to a `Kite Dataset
<http://kitesdk.org/docs/current/guide/>`_.
This sink will deserialize the body of each incoming event and store the
-resulting record in a Kite Dataset. It determines target Dataset by opening a
-repository URI, ``kite.repo.uri``, and loading a Dataset by name,
-``kite.dataset.name``.
+resulting record in a Kite Dataset. It determines target Dataset by loading a
+dataset by URI.
The only supported serialization is avro, and the record schema must be passed
in the event headers, using either ``flume.avro.schema.literal`` with the JSON
@@ -2065,20 +2354,104 @@ Note 2: In some cases, file rolling may
has been exceeded. However, this delay will not exceed 5 seconds. In most
cases, the delay is neglegible.
-======================= =======
===========================================================
-Property Name Default Description
-======================= =======
===========================================================
-**channel** --
-**type** -- Must be
org.apache.flume.sink.kite.DatasetSink
-**kite.repo.uri** -- URI of the repository to open
-**kite.dataset.name** -- Name of the Dataset where records will be
written
-kite.batchSize 100 Number of records to process in each batch
-kite.rollInterval 30 Maximum wait time (seconds) before data
files are released
-auth.kerberosPrincipal -- Kerberos user principal for secure
authentication to HDFS
-auth.kerberosKeytab -- Kerberos keytab location (local FS) for the
principal
-auth.proxyUser -- The effective user for HDFS actions, if
different from
- the kerberos principal
-======================= =======
===========================================================
+============================ =======
===========================================================
+Property Name Default Description
+============================ =======
===========================================================
+**channel** --
+**type** -- Must be
org.apache.flume.sink.kite.DatasetSink
+**kite.dataset.uri** -- URI of the dataset to open
+kite.repo.uri -- URI of the repository to open
+ (deprecated; use kite.dataset.uri
instead)
+kite.dataset.namespace -- Namespace of the Dataset where records
will be written
+ (deprecated; use kite.dataset.uri
instead)
+kite.dataset.name -- Name of the Dataset where records will
be written
+ (deprecated; use kite.dataset.uri
instead)
+kite.batchSize 100 Number of records to process in each
batch
+kite.rollInterval 30 Maximum wait time (seconds) before data
files are released
+kite.flushable.commitOnBatch true If ``true``, the Flume transaction will
be commited and the
+ writer will be flushed on each batch of
``kite.batchSize``
+ records. This setting only applies to
flushable datasets. When
+ ``true``, it's possible for temp files
with commited data to be
+ left in the dataset directory. These
files need to be recovered
+ by hand for the data to be visible to
DatasetReaders.
+kite.syncable.syncOnBatch true Controls whether the sink will also
sync data when committing
+ the transaction. This setting only
applies to syncable datasets.
+ Syncing gaurentees that data will be
written on stable storage
+ on the remote system while flushing
only gaurentees that data
+ has left Flume's client buffers. When
the
+ ``kite.flushable.commitOnBatch``
property is set to ``false``,
+ this property must also be set to
``false``.
+kite.entityParser avro Parser that turns Flume ``Events`` into
Kite entities.
+ Valid values are ``avro`` and the
fully-qualified class name
+ of an implementation of the
``EntityParser.Builder`` interface.
+kite.failurePolicy retry Policy that handles non-recoverable
errors such as a missing
+ ``Schema`` in the ``Event`` header. The
default value, ``retry``,
+ will fail the current batch and try
again which matches the old
+ behavior. Other valid values are
``save``, which will write the
+ raw ``Event`` to the
``kite.error.dataset.uri`` dataset, and the
+ fully-qualified class name of an
implementation of the
+ ``FailurePolicy.Builder`` interface.
+kite.error.dataset.uri -- URI of the dataset where failed events
are saved when
+ ``kite.failurePolicy`` is set to
``save``. **Required** when
+ the ``kite.failurePolicy`` is set to
``save``.
+auth.kerberosPrincipal -- Kerberos user principal for secure
authentication to HDFS
+auth.kerberosKeytab -- Kerberos keytab location (local FS) for
the principal
+auth.proxyUser -- The effective user for HDFS actions, if
different from
+ the kerberos principal
+============================ =======
===========================================================
+
+
+Kafka Sink
+~~~~~~~~~~
+This is a Flume Sink implementation that can publish data to a
+`Kafka <http://kafka.apache.org/>`_ topic. One of the objective is to
integrate Flume
+with Kafka so that pull based processing systems can process the data coming
+through various Flume sources. This currently supports Kafka 0.8.x series of
releases.
+
+Required properties are marked in bold font.
+
+
+=============================== ===================
=============================================================================================
+Property Name Default Description
+=============================== ===================
=============================================================================================
+**type** -- Must be set to
``org.apache.flume.sink.kafka.KafkaSink``
+**brokerList** -- List of brokers
Kafka-Sink will connect to, to get the list of topic partitions
+ This can be a partial
list of brokers, but we recommend at least two for HA.
+ The format is comma
separated list of hostname:port
+topic default-flume-topic The topic in Kafka to
which the messages will be published. If this parameter is configured,
+ messages will be
published to this topic.
+ If the event header
contains a "topic" field, the event will be published to that topic
+ overriding the topic
configured here.
+batchSize 100 How many messages to
process in one batch. Larger batches improve throughput while adding latency.
+requiredAcks 1 How many replicas must
acknowledge a message before its considered successfully written.
+ Accepted values are 0
(Never wait for acknowledgement), 1 (wait for leader only), -1 (wait for all
replicas)
+ Set this to -1 to avoid
data loss in some cases of leader failure.
+Other Kafka Producer Properties -- These properties are
used to configure the Kafka Producer. Any producer property supported
+ by Kafka can be used.
The only requirement is to prepend the property name with the prefix ``kafka.``.
+ For example:
kafka.producer.type
+=============================== ===================
=============================================================================================
+
+.. note:: Kafka Sink uses the ``topic`` and ``key`` properties from the
FlumeEvent headers to send events to Kafka.
+ If ``topic`` exists in the headers, the event will be sent to that
specific topic, overriding the topic configured for the Sink.
+ If ``key`` exists in the headers, the key will used by Kafka to
partition the data between the topic partitions. Events with same key
+ will be sent to the same partition. If the key is null, events
will be sent to random partitions.
+
+An example configuration of a Kafka sink is given below. Properties starting
+with the prefix ``kafka`` (the last 3 properties) are used when instantiating
+the Kafka producer. The properties that are passed when creating the Kafka
+producer are not limited to the properties given in this example.
+Also it's possible include your custom properties here and access them inside
+the preprocessor through the Flume Context object passed in as a method
+argument.
+
+.. code-block:: properties
+
+ a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
+ a1.sinks.k1.topic = mytopic
+ a1.sinks.k1.brokerList = localhost:9092
+ a1.sinks.k1.requiredAcks = 1
+ a1.sinks.k1.batchSize = 20
+ a1.sinks.k1.channel = c1
Custom Sink
~~~~~~~~~~~
@@ -2189,6 +2562,60 @@ Example for agent named a1:
a1.channels = c1
a1.channels.c1.type = jdbc
+Kafka Channel
+~~~~~~~~~~~~~
+
+The events are stored in a Kafka cluster (must be installed separately). Kafka
provides high availability and
+replication, so in case an agent or a kafka broker crashes, the events are
immediately available to other sinks
+
+The Kafka channel can be used for multiple scenarios:
+* With Flume source and sink - it provides a reliable and highly available
channel for events
+* With Flume source and interceptor but no sink - it allows writing Flume
events into a Kafka topic, for use by other apps
+* With Flume sink, but no source - it is a low-latency, fault tolerant way to
send events from Kafka to Flume sources such as HDFS, HBase or Solr
+
+Required properties are in **bold**.
+
+====================== ==========================
===============================================================================================================
+Property Name Default Description
+====================== ==========================
===============================================================================================================
+**type** -- The component type name,
needs to be ``org.apache.flume.channel.kafka.KafkaChannel``
+**brokerList** -- List of brokers in the
Kafka cluster used by the channel
+ This can be a partial list
of brokers, but we recommend at least two for HA.
+ The format is comma
separated list of hostname:port
+**zookeeperConnect** -- URI of ZooKeeper used by
Kafka cluster
+ The format is comma
separated list of hostname:port. If chroot is used, it is added once at the end.
+ For example:
zookeeper-1:2181,zookeeper-2:2182,zookeeper-3:2181/kafka
+topic flume-channel Kafka topic which the
channel will use
+groupId flume Consumer group ID the
channel uses to register with Kafka.
+ Multiple channels must use
the same topic and group to ensure that when one agent fails another can get
the data
+ Note that having
non-channel consumers with the same ID can lead to data loss.
+parseAsFlumeEvent true Expecting Avro datums with
FlumeEvent schema in the channel.
+ This should be true if
Flume source is writing to the channel
+ And false if other
producers are writing into the topic that the channel is using
+ Flume source messages to
Kafka can be parsed outside of Flume by using
+
org.apache.flume.source.avro.AvroFlumeEvent provided by the flume-ng-sdk
artifact
+readSmallestOffset false When set to true, the
channel will read all data in the topic, starting from the oldest event
+ when false, it will read
only events written after the channel started
+ When "parseAsFlumeEvent"
is true, this will be false. Flume source will start prior to the sinks and this
+ guarantees that events
sent by source before sinks start will not be lost.
+Other Kafka Properties -- These properties are used
to configure the Kafka Producer and Consumer used by the channel.
+ Any property supported by
Kafka can be used.
+ The only requirement is to
prepend the property name with the prefix ``kafka.``.
+ For example:
kafka.producer.type
+====================== ==========================
===============================================================================================================
+
+.. note:: Due to the way the channel is load balanced, there may be duplicate
events when the agent first starts up
+
+Example for agent named a1:
+
+.. code-block:: properties
+
+ a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel
+ a1.channels.channel1.capacity = 10000
+ a1.channels.channel1.transactionCapacity = 1000
+ a1.channels.channel1.brokerList=kafka-2:9092,kafka-3:9092
+ a1.channels.channel1.topic=channel1
+ a1.channels.channel1.zookeeperConnect=kafka-1:2181
File Channel
~~~~~~~~~~~~
@@ -2211,6 +2638,7 @@ capacity
keep-alive 3
Amount of time (in sec) to wait for a put operation
use-log-replay-v1 false
Expert: Use old replay logic
use-fast-replay false
Expert: Replay without using queue
+checkpointOnClose true
Controls if a checkpoint is created when the channel is closed. Creating a
checkpoint on close speeds up subsequent startup of the file channel by
avoiding replay.
encryption.activeKey --
Key name used to encrypt new data
encryption.cipherProvider --
Cipher provider type, supported types: AESCTRNOPADDING
encryption.keyProvider --
Key provider type, supported types: JCEKSFILE
@@ -2533,7 +2961,12 @@ that so long as one is available events
The failover mechanism works by relegating failed sinks to a pool where
they are assigned a cool down period, increasing with sequential failures
before they are retried. Once a sink successfully sends an event, it is
-restored to the live pool.
+restored to the live pool. The Sinks have a priority associated with them,
+larger the number, higher the priority. If a Sink fails while sending a Event
+the next Sink with highest priority shall be tried next for sending Events.
+For example, a sink with priority 100 is activated before the Sink with
priority
+80. If no priority is specified, thr priority is determined based on the order
in which
+the Sinks are specified in configuration.
To configure, set a sink groups processor to ``failover`` and set
priorities for all individual sinks. All specified priorities must
@@ -2547,8 +2980,9 @@ Property Name Defau
================================= ===========
===================================================================================
**sinks** -- Space-separated list of sinks
that are participating in the group
**processor.type** ``default`` The component type name, needs
to be ``failover``
-**processor.priority.<sinkName>** -- <sinkName> must be one of
the sink instances associated with the current sink group
-processor.maxpenalty 30000 (in millis)
+**processor.priority.<sinkName>** -- Priority value. <sinkName>
must be one of the sink instances associated with the current sink group
+ A higher priority value Sink
gets activated earlier. A larger absolute value indicates higher priority
+processor.maxpenalty 30000 The maximum backoff period for
the failed Sink (in millis)
================================= ===========
===================================================================================
Example for agent named a1:
@@ -2836,6 +3270,45 @@ Sample flume.conf file:
a1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineFile =
/etc/flume-ng/conf/morphline.conf
a1.sources.avroSrc.interceptors.morphlineinterceptor.morphlineId = morphline1
+Search and Replace Interceptor
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+This interceptor provides simple string-based search-and-replace functionality
+based on Java regular expressions. Backtracking / group capture is also
available.
+This interceptor uses the same rules as in the Java Matcher.replaceAll()
method.
+
+================ =======
========================================================================
+Property Name Default Description
+================ =======
========================================================================
+**type** -- The component type name has to be ``search_replace``
+searchPattern -- The pattern to search for and replace.
+replaceString -- The replacement string.
+charset UTF-8 The charset of the event body. Assumed by default
to be UTF-8.
+================ =======
========================================================================
+
+Example configuration:
+
+.. code-block:: properties
+
+ a1.sources.avroSrc.interceptors = search-replace
+ a1.sources.avroSrc.interceptors.search-replace.type = search_replace
+
+ # Remove leading alphanumeric characters in an event body.
+ a1.sources.avroSrc.interceptors.search-replace.searchPattern = ^[A-Za-z0-9_]+
+ a1.sources.avroSrc.interceptors.search-replace.replaceString =
+
+Another example:
+
+.. code-block:: properties
+
+ a1.sources.avroSrc.interceptors = search-replace
+ a1.sources.avroSrc.interceptors.search-replace.type = search_replace
+
+ # Use grouping operators to reorder and munge words on a line.
+ a1.sources.avroSrc.interceptors.search-replace.searchPattern = The quick
brown ([a-z]+) jumped over the lazy ([a-z]+)
+ a1.sources.avroSrc.interceptors.search-replace.replaceString = The hungry $2
ate the careless $1
+
+
Regex Filtering Interceptor
~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -2947,7 +3420,7 @@ Log4J Appender
Appends Log4j events to a flume agent's avro source. A client using this
appender must have the flume-ng-sdk in the classpath (eg,
-flume-ng-sdk-1.5.2.jar).
+flume-ng-sdk-1.6.0.jar).
Required properties are in **bold**.
===================== =======
==================================================================================
@@ -3011,7 +3484,7 @@ Load Balancing Log4J Appender
Appends Log4j events to a list of flume agent's avro source. A client using
this
appender must have the flume-ng-sdk in the classpath (eg,
-flume-ng-sdk-1.5.2.jar). This appender supports a round-robin and random
+flume-ng-sdk-1.6.0.jar). This appender supports a round-robin and random
scheme for performing the load balancing. It also supports a configurable
backoff
timeout so that down agents are removed temporarily from the set of hosts
Required properties are in **bold**.
@@ -3076,9 +3549,14 @@ Sample log4j.properties file configured
Security
========
-The HDFS sink supports Kerberos authentication if the underlying HDFS is
-running in secure mode. Please refer to the HDFS Sink section for
-configuring the HDFS sink Kerberos-related options.
+The HDFS sink, HBase sink, Thrift source, Thrift sink and Kite Dataset sink
all support
+Kerberos authentication. Please refer to the corresponding sections for
+configuring the Kerberos-related options.
+
+Flume agent will authenticate to the kerberos KDC as a single principal, which
will be
+used by different components that require kerberos authentication. The
principal and
+keytab configured for Thrift source, Thrift sink, HDFS sink, HBase sink and
DataSet sink
+should be the same, otherwise the component will fail to start.
Monitoring
==========
@@ -3087,6 +3565,16 @@ Monitoring in Flume is still a work in p
Several Flume components report metrics to the JMX platform MBean server. These
metrics can be queried using Jconsole.
+JMX Reporting
+-------------
+
+JMX Reporting can be enabled by specifying JMX parameters in the JAVA_OPTS
environment variable using
+flume-env.sh, like
+
+ export JAVA_OPTS="-Dcom.sun.management.jmxremote
-Dcom.sun.management.jmxremote.port=5445
-Dcom.sun.management.jmxremote.authenticate=false
-Dcom.sun.management.jmxremote.ssl=false"
+
+NOTE: The sample above disables the security. To enable Security, please refer
http://docs.oracle.com/javase/6/docs/technotes/guides/management/agent.html
+
Ganglia Reporting
-----------------
Flume can also report these metrics to
@@ -3237,6 +3725,32 @@ metrics as long values.
}
+Tools
+=====
+
+File Channel Integrity Tool
+---------------------------
+
+File Channel Integrity tool verifies the integrity of individual Events in the
File channel
+and removes corrupted Events.
+
+The tools can be run as follows::
+
+ $bin/flume-ng tool --conf ./conf FCINTEGRITYTOOL -l ./datadir
+
+where datadir the comma separated list of data directory to ve verified.
+
+Following are the options available
+
+=======================
====================================================================
+Option Name Description
+=======================
====================================================================
+h/help Displays help
+**l/dataDirs** Comma-separated list of data directories which the
tool must verify
+=======================
====================================================================
+
+
+
Topology Design Considerations
==============================
Modified: websites/staging/flume/trunk/content/_sources/download.txt
==============================================================================
--- websites/staging/flume/trunk/content/_sources/download.txt (original)
+++ websites/staging/flume/trunk/content/_sources/download.txt Mon Jun 1
19:50:47 2015
@@ -10,8 +10,8 @@ originals on the main distribution serve
.. csv-table::
- "Apache Flume binary (tar.gz)", `apache-flume-1.5.2-bin.tar.gz
<http://www.apache.org/dyn/closer.cgi/flume/1.5.2/apache-flume-1.5.2-bin.tar.gz>`_,
`apache-flume-1.5.2-bin.tar.gz.md5
<http://www.us.apache.org/dist/flume/1.5.2/apache-flume-1.5.2-bin.tar.gz.md5>`_,
`apache-flume-1.5.2-bin.tar.gz.asc
<http://www.us.apache.org/dist/flume/1.5.2/apache-flume-1.5.2-bin.tar.gz.asc>`_
- "Apache Flume source (tar.gz)", `apache-flume-1.5.2-src.tar.gz
<http://www.apache.org/dyn/closer.cgi/flume/1.5.2/apache-flume-1.5.2-src.tar.gz>`_,
`apache-flume-1.5.2-src.tar.gz.md5
<http://www.us.apache.org/dist/flume/1.5.2/apache-flume-1.5.2-src.tar.gz.md5>`_,
`apache-flume-1.5.2-src.tar.gz.asc
<http://www.us.apache.org/dist/flume/1.5.2/apache-flume-1.5.2-src.tar.gz.asc>`_
+ "Apache Flume binary (tar.gz)", `apache-flume-1.6.0-bin.tar.gz
<http://www.apache.org/dyn/closer.cgi/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz>`_,
`apache-flume-1.6.0-bin.tar.gz.md5
<http://www.us.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz.md5>`_,
`apache-flume-1.6.0-bin.tar.gz.asc
<http://www.us.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-bin.tar.gz.asc>`_
+ "Apache Flume source (tar.gz)", `apache-flume-1.6.0-src.tar.gz
<http://www.apache.org/dyn/closer.cgi/flume/1.6.0/apache-flume-1.6.0-src.tar.gz>`_,
`apache-flume-1.6.0-src.tar.gz.md5
<http://www.us.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-src.tar.gz.md5>`_,
`apache-flume-1.6.0-src.tar.gz.asc
<http://www.us.apache.org/dist/flume/1.6.0/apache-flume-1.6.0-src.tar.gz.asc>`_
It is essential that you verify the integrity of the downloaded files using
the PGP or MD5 signatures. Please read
`Verifying Apache HTTP Server Releases
<http://httpd.apache.org/dev/verification.html>`_ for more information on
@@ -23,9 +23,9 @@ as well as the asc signature file for th
Then verify the signatures using::
% gpg --import KEYS
- % gpg --verify apache-flume-1.5.2-src.tar.gz.asc
+ % gpg --verify apache-flume-1.6.0-src.tar.gz.asc
-Apache Flume 1.5.2 is signed by Hari Shreedharan 77FFC9AB
+Apache Flume 1.6.0 is signed by Johny Rufus 2C79120F
Alternatively, you can verify the MD5 or SHA1 signatures of the files. A
program called md5, md5sum, or shasum is included in many
Unix distributions for this purpose.
Modified: websites/staging/flume/trunk/content/_sources/index.txt
==============================================================================
--- websites/staging/flume/trunk/content/_sources/index.txt (original)
+++ websites/staging/flume/trunk/content/_sources/index.txt Mon Jun 1 19:50:47
2015
@@ -33,6 +33,38 @@ application.
.. raw:: html
+ <h3>May 20, 2015 - Apache Flume 1.6.0 Released</h3>
+
+The Apache Flume team is pleased to announce the release of Flume 1.6.0.
+
+Flume is a distributed, reliable, and available service for efficiently
+collecting, aggregating, and moving large amounts of streaming event data.
+
+Version 1.6.0 is the ninth Flume release as an Apache top-level project.
+Flume 1.6.0 is stable, production-ready software, and is backwards-compatible
+with previous versions of the Flume 1.x codeline.
+
+Several months of active development went into this release: 105 patches were
committed since 1.5.2, representing many features, enhancements, and bug fixes.
While the full change log can be found on the 1.6.0 release page (link below),
here are a few new feature highlights:
+
+ * Flume Sink and Source for Apache Kafka
+ * A new channel that uses Kafka
+ * Hive Sink based on the new Hive Streaming support
+ * End to End authentication in Flume
+ * Simple regex search-and-replace interceptor
+
+The full change log and documentation are available on the
+`Flume 1.6.0 release page <releases/1.6.0.html>`__.
+
+This release can be downloaded from the Flume `Download <download.html>`__
page.
+
+Your contributions, feedback, help and support make Flume better!
+For more information on how to report problems or contribute,
+please visit our `Get Involved <getinvolved.html>`__ page.
+
+The Apache Flume Team
+
+.. raw:: html
+
<h3>November 18, 2014 - Apache Flume 1.5.2 Released</h3>
The Apache Flume team is pleased to announce the release of Flume 1.5.2
