Flume Sinks

HDFS Sink

This sink writes events into the Hadoop Distributed File System (HDFS). Itcurrently supports creating text and sequence files. It supports compression inboth file types. The files can be rolled (close current file and create a newone) periodically based on the elapsed time or size of data or number of events.It also buckets/partitions data by attributes like timestamp or machinewhere the event originated. The HDFS directory path may contain formattingescape sequences that will replaced by the HDFS sink to generate adirectory/file name to store the events. Using this sink requires hadoop to beinstalled so that Flume can use the Hadoop jars to communicate with the HDFScluster. Note that a version of Hadoop that supports the sync() call isrequired.

The following are the escape sequences supported:

Note: The escape strings %[localhost], %[IP] and %[FQDN] all rely on Java’s ability to obtain thehostname, which may fail in some networking environments.

The file in use will have the name mangled to include ”.tmp” at the end. Oncethe file is closed, this extension is removed. This allows excluding partiallycomplete files in the directory.Required properties are in bold.

Note

For all of the time related escape sequences, a header with the key“timestamp” must exist among the headers of the event (unless hdfs.useLocalTimeStamp is set to true). One way to addthis automatically is to use the TimestampInterceptor.

Deprecated Properties

Name Default Description====================== ============ ======================================================================hdfs.callTimeout 30000 Number of milliseconds allowed for HDFS operations, such as open, write, flush, close. This number should be increased if many HDFS timeout operations are occurring.====================== ============ ======================================================================

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute

The above configuration will round down the timestamp to the last 10th minute. For example, an event withtimestamp 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 becomeimmediately visible to Hive queries. Partitions to which flume will stream to can either be pre-createdor, optionally, Flume can create them if they are missing. Fields from incoming event data are mapped tocorresponding columns in the Hive table.

Following serializers are provided for Hive sink:

JSON: Handles UTF8 encoded Json (strict syntax) events and requires no configration. Object namesin 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.

The following are the escape sequences supported:

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 addthis automatically is to use the TimestampInterceptor.

Example Hive table :

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:

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 withtimestamp header set to 11:54:34 AM, June 12, 2012 and ‘country’ header set to ‘india’ will evaluate to thepartition (continent=’asia’,country=’india’,time=‘2012-06-12-11-50’. The serializer is configured toaccept tab separated input containing three fields and to skip the second field.

Logger Sink

Logs event at INFO level. Typically useful for testing/debugging purpose. Required properties arein bold. This sink is the only exception which doesn’t require the extra configurationexplained in the Logging raw data section.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = logger
a1.sinks.k1.channel = c1

Avro Sink

This sink forms one half of Flume’s tiered collection support. Flume eventssent to this sink are turned into Avro events and sent to the configuredhostname / port pair. The events are taken from the configured Channel inbatches of the configured batch size.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = avro
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 10.10.10.10
a1.sinks.k1.port = 4545

Thrift Sink

This sink forms one half of Flume’s tiered collection support. Flume eventssent to this sink are turned into Thrift events and sent to the configuredhostname / port pair. The events are taken from the configured Channel inbatches 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 alsooperate in secure mode. client-principal and client-keytab are the properties used by theThrift sink to authenticate to the kerberos KDC. The server-principal represents theprincipal of the Thrift source this sink is configured to connect to in secure mode.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = thrift
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 10.10.10.10
a1.sinks.k1.port = 4545

IRC Sink

The IRC sink takes messages from attached channel and relays those toconfigured IRC destinations.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = irc
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = irc.yourdomain.com
a1.sinks.k1.nick = flume
a1.sinks.k1.chan = #flume

File Roll Sink

Stores events on the local filesystem.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = file_roll
a1.sinks.k1.channel = c1
a1.sinks.k1.sink.directory = /var/log/flume

Null Sink

Discards all events it receives from the channel.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = null
a1.sinks.k1.channel = c1

HBaseSinks

HBaseSink

This sink writes data to HBase. The Hbase configuration is picked up from the firsthbase-site.xml encountered in the classpath. A class implementing HbaseEventSerializerwhich is specified by the configuration is used to convert the events intoHBase puts and/or increments. These puts and increments are then writtento HBase. This sink provides the same consistency guarantees as HBase,which is currently row-wise atomicity. In the event of Hbase failing towrite certain events, the sink will replay all events in that transaction.

The HBaseSink supports writing data to secure HBase. To write to secure HBase, the userthe agent is running as must have write permissions to the table the sink is configuredto write to. The principal and keytab to use to authenticate against the KDC can be specifiedin the configuration. The hbase-site.xml in the Flume agent’s classpathmust have authentication set to kerberos (For details on how to do this, please refer toHBase documentation).

For convenience, two serializers are provided with Flume. TheSimpleHbaseEventSerializer (org.apache.flume.sink.hbase.SimpleHbaseEventSerializer)writes the event bodyas-is to HBase, and optionally increments a column in Hbase. This is primarilyan example implementation. The RegexHbaseEventSerializer(org.apache.flume.sink.hbase.RegexHbaseEventSerializer) breaks the event bodybased on the given regex and writes each part into different columns.

The type is the FQCN: org.apache.flume.sink.hbase.HBaseSink.

Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hbase
a1.sinks.k1.table = foo_table
a1.sinks.k1.columnFamily = bar_cf
a1.sinks.k1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer
a1.sinks.k1.channel = c1

HBase2Sink

HBase2Sink is the equivalent of HBaseSink for HBase version 2.The provided functionality and the configuration parameters are the same as in case of HBaseSink (except the hbase2 tag in the sink type and the package/class names).

The type is the FQCN: org.apache.flume.sink.hbase2.HBase2Sink.

Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hbase2
a1.sinks.k1.table = foo_table
a1.sinks.k1.columnFamily = bar_cf
a1.sinks.k1.serializer = org.apache.flume.sink.hbase2.RegexHBase2EventSerializer
a1.sinks.k1.channel = c1

AsyncHBaseSink

This sink writes data to HBase using an asynchronous model. A class implementingAsyncHbaseEventSerializer which is specified by the configuration is used to convert the events intoHBase puts and/or increments. These puts and increments are then writtento HBase. This sink uses the Asynchbase API to write toHBase. This sink provides the same consistency guarantees as HBase,which is currently row-wise atomicity. In the event of Hbase failing towrite certain events, the sink will replay all events in that transaction.AsyncHBaseSink can only be used with HBase 1.x. The async client library used by AsyncHBaseSink is not available for HBase 2.The type is the FQCN: org.apache.flume.sink.hbase.AsyncHBaseSink.Required properties are in bold.

Note that this sink takes the Zookeeper Quorum and parent znode information inthe configuration. Zookeeper Quorum and parent node configuration may bespecified in the flume configuration file. Alternatively, these configurationvalues are taken from the first hbase-site.xml file in the classpath.

If these are not provided in the configuration, then the sinkwill read this information from the first hbase-site.xml file in the classpath.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = asynchbase
a1.sinks.k1.table = foo_table
a1.sinks.k1.columnFamily = bar_cf
a1.sinks.k1.serializer = org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer
a1.sinks.k1.channel = c1

MorphlineSolrSink

This sink extracts data from Flume events, transforms it, and loads it in near-real-time into Apache Solr servers, which in turn serve queries to end users or search applications.

This sink is well suited for use cases that stream raw data into HDFS (via the HdfsSink) and simultaneously extract, transform and load the same data into Solr (via MorphlineSolrSink). In particular, this sink can process arbitrary heterogeneous raw data from disparate data sources and turn it into a data model that is useful to Search applications.

The ETL functionality is customizable using a morphline configuration file that defines a chain of transformation commands that pipe event records from one command to another.

Morphlines can be seen as an evolution of Unix pipelines where the data model is generalized to work with streams of generic records, including arbitrary binary payloads. A morphline command is a bit like a Flume Interceptor. Morphlines can be embedded into Hadoop components such as Flume.

Commands to parse and transform a set of standard data formats such as log files, Avro, CSV, Text, HTML, XML, PDF, Word, Excel, etc. are provided out of the box, and additional custom commands and parsers for additional data formats can be added as morphline plugins. Any kind of data format can be indexed and any Solr documents for any kind of Solr schema can be generated, and any custom ETL logic can be registered and executed.

Morphlines manipulate continuous streams of records. The data model can be described as follows: A record is a set of named fields where each field has an ordered list of one or more values. A value can be any Java Object. That is, a record is essentially a hash table where each hash table entry contains a String key and a list of Java Objects as values. (The implementation uses Guava’s ArrayListMultimap, which is a ListMultimap). Note that a field can have multiple values and any two records need not use common field names.

This sink fills the body of the Flume event into the _attachment_body field of the morphline record, as well as copies the headers of the Flume event into record fields of the same name. The commands can then act on this data.

Routing to a SolrCloud cluster is supported to improve scalability. Indexing load can be spread across a large number of MorphlineSolrSinks for improved scalability. Indexing load can be replicated across multiple MorphlineSolrSinks for high availability, for example using Flume features such as Load balancing Sink Processor. MorphlineInterceptor can also help to implement dynamic routing to multiple Solr collections (e.g. for multi-tenancy).

The morphline and solr jars required for your environment must be placed in the lib directory of the Apache Flume installation.

The type is the FQCN: org.apache.flume.sink.solr.morphline.MorphlineSolrSink

Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = org.apache.flume.sink.solr.morphline.MorphlineSolrSink
a1.sinks.k1.channel = c1
a1.sinks.k1.morphlineFile = /etc/flume-ng/conf/morphline.conf
# a1.sinks.k1.morphlineId = morphline1
# a1.sinks.k1.batchSize = 1000
# a1.sinks.k1.batchDurationMillis = 1000

ElasticSearchSink

This sink writes data to an elasticsearch cluster. By default, events will be written so that the Kibana graphical interfacecan display them - just as if logstash wrote them.

The elasticsearch and lucene-core jars required for your environment must be placed in the lib directory of the Apache Flume installation.Elasticsearch requires that the major version of the client JAR match that of the server and that both are running the same minor versionof the JVM. SerializationExceptions will appear if this is incorrect. Toselect the required version first determine the version of elasticsearch and the JVM version the target cluster is running. Then select an elasticsearch clientlibrary which matches the major version. A 0.19.x client can talk to a 0.19.x cluster; 0.20.x can talk to 0.20.x and 0.90.x can talk to 0.90.x. Once theelasticsearch version has been determined then read the pom.xml file to determine the correct lucene-core JAR version to use. The Flume agentwhich is running the ElasticSearchSink should also match the JVM the target cluster is running down to the minor version.

Events will be written to a new index every day. The name will be <indexName>-yyyy-MM-dd where <indexName> is the indexName parameter. The sinkwill start writing to a new index at midnight UTC.

Events are serialized for elasticsearch by the ElasticSearchLogStashEventSerializer by default. This behaviour can beoverridden with the serializer parameter. This parameter accepts implementations of org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializeror org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory. Implementing ElasticSearchEventSerializer is deprecated in favour ofthe more powerful ElasticSearchIndexRequestBuilderFactory.

The type is the FQCN: org.apache.flume.sink.elasticsearch.ElasticSearchSink

Required properties are in bold.

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:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = elasticsearch
a1.sinks.k1.hostNames = 127.0.0.1:9200,127.0.0.2:9300
a1.sinks.k1.indexName = foo_index
a1.sinks.k1.indexType = bar_type
a1.sinks.k1.clusterName = foobar_cluster
a1.sinks.k1.batchSize = 500
a1.sinks.k1.ttl = 5d
a1.sinks.k1.serializer = org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
a1.sinks.k1.channel = c1

Kite Dataset Sink

Experimental sink that writes events to a Kite Dataset.This sink will deserialize the body of each incoming event and store theresulting record in a Kite Dataset. It determines target Dataset by loading adataset by URI.

The only supported serialization is avro, and the record schema must be passedin the event headers, using either flume.avro.schema.literal with the JSONschema representation or flume.avro.schema.url with a URL where the schemamay be found (hdfs:/… URIs are supported). This is compatible with theLog4jAppender flume client and the spooling directory source’s Avrodeserializer using deserializer.schemaType = LITERAL.

Note 1: The flume.avro.schema.hash header is not supported.Note 2: In some cases, file rolling may occur slightly after the roll intervalhas been exceeded. However, this delay will not exceed 5 seconds. In mostcases, the delay is neglegible.

Kafka Sink

This is a Flume Sink implementation that can publish data to aKafka topic. One of the objective is to integrate Flumewith Kafka so that pull based processing systems can process the data comingthrough various Flume sources.

This currently supports Kafka server releases 0.10.1.0 or higher. Testing was done up to 2.0.1 that was the highest avilable version at the time of the release.

Required properties are marked in bold font.

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 keywill be sent to the same partition. If the key is null, events will be sent to random partitions.

The Kafka sink also provides defaults for the key.serializer(org.apache.kafka.common.serialization.StringSerializer)and value.serializer(org.apache.kafka.common.serialization.ByteArraySerializer). Modification of these parameters is not recommended.

Deprecated Properties

An example configuration of a Kafka sink is given below. Properties startingwith the prefix kafka.producer the Kafka producer. The properties that are passed when creating the Kafkaproducer are not limited to the properties given in this example.Also it is possible to include your custom properties here and access them insidethe preprocessor through the Flume Context object passed in as a methodargument.

a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy

Security and Kafka Sink:

Secure authentication as well as data encryption is supported on the communication channel between Flume and Kafka.For secure authentication SASL/GSSAPI (Kerberos V5) or SSL (even though the parameter is named SSL, the actual protocol is a TLS implementation) can be used from Kafka version 0.9.0.

As of now data encryption is solely provided by SSL/TLS.

Setting kafka.producer.security.protocol to any of the following value means:

• SASL_PLAINTEXT - Kerberos or plaintext authentication with no data encryption
• SASL_SSL - Kerberos or plaintext authentication with data encryption
• SSL - TLS based encryption with optional authentication.

Warning

There is a performance degradation when SSL is enabled,the magnitude of which depends on the CPU type and the JVM implementation.Reference: Kafka security overviewand the jira for tracking this issue:KAFKA-2561

TLS and Kafka Sink:

Please read the steps described in Configuring Kafka Clients SSLto learn about additional configuration settings for fine tuning for example any of the following:security provider, cipher suites, enabled protocols, truststore or keystore types.

Example configuration with server side authentication and data encryption.

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.sink1.kafka.bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093
a1.sinks.sink1.kafka.topic = mytopic
a1.sinks.sink1.kafka.producer.security.protocol = SSL
# optional, the global truststore can be used alternatively
a1.sinks.sink1.kafka.producer.ssl.truststore.location = /path/to/truststore.jks
a1.sinks.sink1.kafka.producer.ssl.truststore.password = <password to access the truststore>

Specyfing the truststore is optional here, the global truststore can be used instead.For more details about the global SSL setup, see the SSL/TLS support section.

Note: By default the property ssl.endpoint.identification.algorithmis not defined, so hostname verification is not performed.In order to enable hostname verification, set the following properties

a1.sinks.sink1.kafka.producer.ssl.endpoint.identification.algorithm = HTTPS

Once enabled, clients will verify the server’s fully qualified domain name (FQDN)against one of the following two fields:

• Common Name (CN) https://tools.ietf.org/html/rfc6125#section-2.3
• Subject Alternative Name (SAN) https://tools.ietf.org/html/rfc5280#section-4.2.1.6
  If client side authentication is also required then additionally the following needs to be added to Flume agentconfiguration or the global SSL setup can be used (see SSL/TLS support section).Each Flume agent has to have its client certificate which has to be trusted by Kafka brokers eitherindividually or by their signature chain. Common example is to sign each client certificate by a single Root CAwhich in turn is trusted by Kafka brokers.

# optional, the global keystore can be used alternatively
a1.sinks.sink1.kafka.producer.ssl.keystore.location = /path/to/client.keystore.jks
a1.sinks.sink1.kafka.producer.ssl.keystore.password = <password to access the keystore>

If keystore and key use different password protection then ssl.key.password property willprovide the required additional secret for producer keystore:

a1.sinks.sink1.kafka.producer.ssl.key.password = <password to access the key>

Kerberos and Kafka Sink:

To use Kafka sink with a Kafka cluster secured with Kerberos, set the producer.security.protocol property noted above for producer.The Kerberos keytab and principal to be used with Kafka brokers is specified in a JAAS file’s “KafkaClient” section. “Client” section describes the Zookeeper connection if needed.See Kafka docfor information on the JAAS file contents. The location of this JAAS file and optionally the system wide kerberos configuration can be specified via JAVA_OPTS in flume-env.sh:

JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/path/to/krb5.conf"
JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/path/to/flume_jaas.conf"

Example secure configuration using SASL_PLAINTEXT:

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.sink1.kafka.bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093
a1.sinks.sink1.kafka.topic = mytopic
a1.sinks.sink1.kafka.producer.security.protocol = SASL_PLAINTEXT
a1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPI
a1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka

Example secure configuration using SASL_SSL:

a1.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.sink1.kafka.bootstrap.servers = kafka-1:9093,kafka-2:9093,kafka-3:9093
a1.sinks.sink1.kafka.topic = mytopic
a1.sinks.sink1.kafka.producer.security.protocol = SASL_SSL
a1.sinks.sink1.kafka.producer.sasl.mechanism = GSSAPI
a1.sinks.sink1.kafka.producer.sasl.kerberos.service.name = kafka
# optional, the global truststore can be used alternatively
a1.sinks.sink1.kafka.producer.ssl.truststore.location = /path/to/truststore.jks
a1.sinks.sink1.kafka.producer.ssl.truststore.password = <password to access the truststore>

Sample JAAS file. For reference of its content please see client config sections of the desired authentication mechanism (GSSAPI/PLAIN)in Kafka documentation of SASL configuration.Unlike the Kafka Source or Kafka Channel a “Client” section is not required, unless it is needed by other connecting components. Also please make surethat the operating system user of the Flume processes has read privileges on the jaas and keytab files.

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  keyTab="/path/to/keytabs/flume.keytab"
  principal="flume/flumehost1.example.com@YOURKERBEROSREALM";
};

HTTP Sink

Behaviour of this sink is that it will take events from the channel, andsend those events to a remote service using an HTTP POST request. The eventcontent is sent as the POST body.

Error handling behaviour of this sink depends on the HTTP response returnedby the target server. The sink backoff/ready status is configurable, as is thetransaction commit/rollback result and whether the event contributes to thesuccessful event drain count.

Any malformed HTTP response returned by the server where the status code isnot readable will result in a backoff signal and the event is not consumedfrom the channel.

Required properties are in bold.

Note that the most specific HTTP status code match is used for the backoff,rollback and incrementMetrics configuration options. If there are configurationvalues for both 2XX and 200 status codes, then 200 HTTP codes will use the 200value, and all other HTTP codes in the 201-299 range will use the 2XX value.

Any empty or null events are consumed without any request being made to theHTTP endpoint.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = http
a1.sinks.k1.channel = c1
a1.sinks.k1.endpoint = http://localhost:8080/someuri
a1.sinks.k1.connectTimeout = 2000
a1.sinks.k1.requestTimeout = 2000
a1.sinks.k1.acceptHeader = application/json
a1.sinks.k1.contentTypeHeader = application/json
a1.sinks.k1.defaultBackoff = true
a1.sinks.k1.defaultRollback = true
a1.sinks.k1.defaultIncrementMetrics = false
a1.sinks.k1.backoff.4XX = false
a1.sinks.k1.rollback.4XX = false
a1.sinks.k1.incrementMetrics.4XX = true
a1.sinks.k1.backoff.200 = false
a1.sinks.k1.rollback.200 = false
a1.sinks.k1.incrementMetrics.200 = true

Custom Sink

A custom sink is your own implementation of the Sink interface. A customsink’s class and its dependencies must be included in the agent’s classpathwhen starting the Flume agent. The type of the custom sink is its FQCN.Required properties are in bold.

Example for agent named a1:

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = org.example.MySink
a1.sinks.k1.channel = c1