大数据技术之 Flume3

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
# Describe/configure the source
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141
# Describe the sink
a2.sinks.k1.type = logger
# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a3 --conf-file job/group2/flume-flume-console2.conf -
Dflume.root.logger=INFO,console
[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a2 --conf-file job/group2/flume-flume-console1.conf -
Dflume.root.logger=INFO,console
[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a1 --conf-file job/group2/flume-netcat-flume.conf

[atguigu@hadoop102 job]$ mkdir group3
[atguigu@hadoop103 job]$ mkdir group3
[atguigu@hadoop104 job]$ mkdir group3

[atguigu@hadoop102 group3]$ vim flume1-logger-flume.conf 
添加如下内容
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /opt/module/group.log
a1.sources.r1.shell = /bin/bash -c
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop104
a1.sinks.k1.port = 4141 
# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c1
# Describe/configure the source
a3.sources.r1.type = avro 
a3.sources.r1.bind = hadoop104
a3.sources.r1.port = 4141
# Describe the sink
# Describe the sink
a3.sinks.k1.type = logger
# Describe the channel
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r1.channels = c1
a3.sinks.k1.channel = c1

<dependency>
     <groupId>org.apache.flume</groupId>
     <artifactId>flume-ng-core</artifactId>
     <version>1.9.0</version>
</dependency>

package com.atguigu.interceptor;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
public class TypeInterceptor implements Interceptor {
     //声明一个存放事件的集合
     private List<Event> addHeaderEvents;
     @Override  
     public void initialize() {
         //初始化存放事件的集合
         addHeaderEvents = new ArrayList<>();
     }
     //单个事件拦截
     @Override
     public Event intercept(Event event) {
         //1.获取事件中的头信息
         Map<String, String> headers = event.getHeaders();
         //2.获取事件中的 body 信息
         String body = new String(event.getBody());
         //3.根据 body 中是否有"atguigu"来决定添加怎样的头信息
         if (body.contains("atguigu")) {
             //4.添加头信息
             headers.put("type", "first");
         } else {
             //4.添加头信息
             headers.put("type", "second");
         }
         return event;
     }
     //批量事件拦截
     @Override
     public List<Event> intercept(List<Event> events) {
         //1.清空集合
         addHeaderEvents.clear();
         //2.遍历 events
         for (Event event : events) {
             //3.给每一个事件添加头信息
             addHeaderEvents.add(intercept(event));
         }
         //4.返回结果
         return addHeaderEvents;
     }
     @Override
     public void close() {
     }
     public static class Builder implements Interceptor.Builder {
         @Override
         public Interceptor build() {
             return new TypeInterceptor();
         }
         @Override 
         public void configure(Context context) {
         }
     }
}