7.Flume Interceptors

灰太狼 2022-06-14 01:40 192阅读 0赞

### Flume Interceptors ###

Flume has the capability to modify/drop events in-flight. This is done with the help of interceptors. Interceptors are classes that implement org.apache.flume.interceptor.Interceptor interface. An interceptor can modify or even drop events based on any criteria chosen by the developer of the interceptor. Flume supports chaining of interceptors. This is made possible through by specifying the list of interceptor builder class names in the configuration. Interceptors are specified as a whitespace separated list in the source configuration. The order in which the interceptors are specified is the order in which they are invoked. The list of events returned by one interceptor is passed to the next interceptor in the chain. Interceptors can modify or drop events. If an interceptor needs to drop events, it just does not return that event in the list that it returns. If it is to drop all events, then it simply returns an empty list. Interceptors are named components, here is an example of how they are created through configuration:

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    a1.sources.r1.interceptors = i1 i2
    a1.sources.r1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Builder
    a1.sources.r1.interceptors.i1.preserveExisting = false
    a1.sources.r1.interceptors.i1.hostHeader = hostname
    a1.sources.r1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder
    a1.sinks.k1.filePrefix = FlumeData.%{CollectorHost}.%Y-%m-%d
    a1.sinks.k1.channel = c1

Note that the interceptor builders are passed to the type config parameter. The interceptors are themselves configurable and can be passed configuration values just like they are passed to any other configurable component. In the above example, events are passed to the HostInterceptor first and the events returned by the HostInterceptor are then passed along to the TimestampInterceptor. You can specify either the fully qualified class name (FQCN) or the alias timestamp. If you have multiple collectors writing to the same HDFS path, then you could also use the HostInterceptor.

#### Timestamp Interceptor ####

This interceptor inserts into the event headers, the time in millis at which it processes the event. This interceptor inserts a header with key timestamp whose value is the relevant timestamp. This interceptor can preserve an existing timestamp if it is already present in the configuration.

Example for agent named a1:

a1.sources = r1
    a1.channels = c1
    a1.sources.r1.channels =  c1
    a1.sources.r1.type = seq
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = timestamp

#### Host Interceptor ####

This interceptor inserts the hostname or IP address of the host that this agent is running on. It inserts a header with key host or a configured key whose value is the hostname or IP address of the host, based on configuration.

Example for agent named a1:

a1.sources = r1
    a1.channels = c1
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = host
    a1.sources.r1.interceptors.i1.hostHeader = hostname

#### Static Interceptor ####

Static interceptor allows user to append a static header with static value to all events.

The current implementation does not allow specifying multiple headers at one time. Instead user might chain multiple static interceptors each defining one static header.

Example for agent named a1:

a1.sources = r1
    a1.channels = c1
    a1.sources.r1.channels =  c1
    a1.sources.r1.type = seq
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = static
    a1.sources.r1.interceptors.i1.key = datacenter
    a1.sources.r1.interceptors.i1.value = NEW_YORK

#### UUID Interceptor ####

This interceptor sets a universally unique identifier on all events that are intercepted. An example UUID is b5755073-77a9-43c1-8fad-b7a586fc1b97, which represents a 128-bit value.

Consider using UUIDInterceptor to automatically assign a UUID to an event if no application level unique key for the event is available. It can be important to assign UUIDs to events as soon as they enter the Flume network; that is, in the first Flume Source of the flow. This enables subsequent deduplication of events in the face of replication and redelivery in a Flume network that is designed for high availability and high performance. If an application level key is available, this is preferable over an auto-generated UUID because it enables subsequent updates and deletes of event in data stores using said well known application level key.

#### Morphline Interceptor ####

This interceptor filters the events through a [morphline configuration file][] that defines a chain of transformation commands that pipe records from one command to another. For example the morphline can ignore certain events or alter or insert certain event headers via regular expression based pattern matching, or it can auto-detect and set a MIME type via Apache Tika on events that are intercepted. For example, this kind of packet sniffing can be used for content based dynamic routing in a Flume topology. MorphlineInterceptor can also help to implement dynamic routing to multiple Apache Solr collections (e.g. for multi-tenancy).

Currently, there is a restriction in that the morphline of an interceptor must not generate more than one output record for each input event. This interceptor is not intended for heavy duty ETL processing - if you need this consider moving ETL processing from the Flume Source to a Flume Sink, e.g. to a MorphlineSolrSink.

Required properties are in **bold**.

Sample flume.conf file:

a1.sources.avroSrc.interceptors = morphlineinterceptor
    a1.sources.avroSrc.interceptors.morphlineinterceptor.type = org.apache.flume.sink.solr.morphline.MorphlineInterceptor$Builder
    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.

Example configuration:

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:

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 ####

This interceptor filters events selectively by interpreting the event body as text and matching the text against a configured regular expression. The supplied regular expression can be used to include events or exclude events.

#### Regex Extractor Interceptor ####

This interceptor extracts regex match groups using a specified regular expression and appends the match groups as headers on the event. It also supports pluggable serializers for formatting the match groups before adding them as event headers.

<table style="border:0px; border-collapse:collapse">      
 <thead> 
  <tr> 
   <th style="text-align:left; padding:1px 8px 1px 5px; background-color:rgb(238,221,238); border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Property Name</th> 
   <th style="text-align:left; padding:1px 8px 1px 5px; background-color:rgb(238,221,238); border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Default</th> 
   <th style="text-align:left; padding:1px 8px 1px 5px; background-color:rgb(238,221,238); border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> <strong>type</strong></td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> –</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> The component type name has to be&nbsp;<span>regex_extractor</span></td> 
  </tr> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> <strong>regex</strong></td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> –</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Regular expression for matching against events</td> 
  </tr> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> <strong>serializers</strong></td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> –</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Space-separated list of serializers for mapping matches to header names and serializing their values. (See example below) Flume provides built-in support for the following serializers:<span>org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer</span><span>org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer</span></td> 
  </tr> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> serializers.&lt;s1&gt;.type</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> default</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Must be&nbsp;<span>default</span>&nbsp;(org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer),<span>org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer</span>, or the FQCN of a custom class that implements&nbsp;<span>org.apache.flume.interceptor.RegexExtractorInterceptorSerializer</span></td> 
  </tr> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> serializers.&lt;s1&gt;.<strong>name</strong></td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> –</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> &nbsp;</td> 
  </tr> 
  <tr> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> serializers.*</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> –</td> 
   <td style="padding:1px 8px 1px 5px; border-width:0px 0px 1px; border-top-style:initial; border-top-color:initial; border-left-style:initial; border-left-color:initial; border-right-style:initial; border-right-color:initial; border-bottom-style:solid; border-bottom-color:rgb(170,170,170)"> Serializer-specific properties</td> 
  </tr> 
 </tbody> 
</table>

The serializers are used to map the matches to a header name and a formatted header value; by default, you only need to specify the header name and the default org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer will be used. This serializer simply maps the matches to the specified header name and passes the value through as it was extracted by the regex. You can plug custom serializer implementations into the extractor using the fully qualified class name (FQCN) to format the matches in anyway you like.

#### Example 1: ####

If the Flume event body contained 1:2:3.4foobar5 and the following configuration was used

a1.sources.r1.interceptors.i1.regex = (\\d):(\\d):(\\d)
    a1.sources.r1.interceptors.i1.serializers = s1 s2 s3
    a1.sources.r1.interceptors.i1.serializers.s1.name = one
    a1.sources.r1.interceptors.i1.serializers.s2.name = two
    a1.sources.r1.interceptors.i1.serializers.s3.name = three

The extracted event will contain the same body but the following headers will have been added one=>1, two=>2, three=>3

#### Example 2: ####

If the Flume event body contained 2012-10-18 18:47:57,614 some log line and the following configuration was used

a1.sources.r1.interceptors.i1.regex = ^(?:\\n)?(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)
    a1.sources.r1.interceptors.i1.serializers = s1
    a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer
    a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp
    a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm

the extracted event will contain the same body but the following headers will have been added timestamp=>1350611220000

[morphline configuration file]: http://cloudera.github.io/cdk/docs/current/cdk-morphlines/index.html