flume对接kafka

梦里梦外; 2023-07-01 14:00 2阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMDQ5Nzcz_size_16_color_FFFFFF_t_70][]

> # flume agent配置 #
> 
>     #定义
>     a1.sources = r1
>     a1.channels = c1 c2
>     a1.sinks = k1 k2
>      
>     #source
>     a1.sources.r1.type = TAILDIR
>     # 用于断点续传，文件中包含各个路径下各个文件当前读取到的偏移量
>     a1.sources.r1.positionFile = /opt/flume_conf/tmp/tail_dir.json
>     # 文件组  如两目录 f1 f2
>     a1.sources.r1.filegroups = f1
>     a1.sources.r1.filegroups.f1 = /opt/data/.*.log
>      
>     #channel
>     a1.channels.c1.type = memory
>     a1.channels.c1.capacity = 1000
>     a1.channels.c1.transactionCapacity = 100
>      
>     a1.channels.c2.type = memory
>     a1.channels.c2.capacity = 1000
>     a1.channels.c2.transactionCapacity = 100
>      
>     # channel selector 定义channel选择器
>     a1.sources.r1.selector.type = multiplexing
>     a1.sources.r1.selector.header = topic
>     a1.sources.r1.selector.mapping.mytopic1 = c1
>     a1.sources.r1.selector.mapping.mytopic2 = c2
>     #拦截器拦截掉的数据可以定义到c3
>     #a1.sources.r1.selector.default = c3
>      
>     #Interceptor 配置自定义拦截器
>     a1.sources.r1.interceptors = i1
>     a1.sources.r1.interceptors.i1.type = com.flume.interceptors.DistributeLogsInterceptor$Builder
>      
>      
>     #sink
>     a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
>     a1.sinks.k1.kafka.bootstrap.servers =node1:9092,node2:9092,node3:9092
>     a1.sinks.k1.kafka.topic = mytopic1
>     a1.sinks.k1.kafka.flumeBatchSize = 20
>     a1.sinks.k1.kafka.producer.acks = 1
>     
>     a1.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink
>     a1.sinks.k2.kafka.bootstrap.servers =node1:9092,node2:9092,node3:9092
>     a1.sinks.k2.kafka.topic = mytopic1
>     a1.sinks.k2.kafka.flumeBatchSize = 20
>     a1.sinks.k2.kafka.producer.acks = 1
>      
>     #bind 关联关系
>     a1.sources.r1.channels = c1 c2
>     a1.sinks.k1.channel = c1
>     a1.sinks.k2.channel = c2
> 
> # flume自定义拦截器，对source数据进行分发 #
> 
>     package com.flume.interceptors;
>     
>     import org.apache.flume.Context;
>     import org.apache.flume.Event;
>     import org.apache.flume.interceptor.Interceptor;
>     import org.apache.velocity.runtime.directive.Foreach;
>     
>     import java.util.ArrayList;
>     import java.util.List;
>     import java.util.Map;
>     
>     /**
>      * Created by HP on 2020/1/17.
>      */
>     public class DistributeLogsInterceptor implements Interceptor{
>         //声明一个添加了头信息的event容器
>         private List<Event> addHeaderEventList;
>     
>         @Override
>         public void initialize() {
>             addHeaderEventList=new ArrayList<>();
>         }
>         /*a1.sources = r1
>         a1.channels = c1 c2 c3 c4
>         a1.sources.r1.selector.type = multiplexing
>         a1.sources.r1.selector.header = state
>         a1.sources.r1.selector.mapping.CZ = c1
>         a1.sources.r1.selector.mapping.US = c2 c3
>         a1.sources.r1.selector.default = c4*/
>         //单个事件的拦截器
>         @Override
>         public Event intercept(Event event) {
>             //获取事件中的头信息
>             // a1.sources.r1.selector.header = state
>             Map<String, String> headers = event.getHeaders();
>             //获取事件中的body信息
>             String body=new String(event.getBody());
>             if(body.contains("PVUV")){//前台页面访问日志
>                 //添加头信息
>                 //a1.sources.r1.selector.header = state
>                 //a1.sources.r1.selector.mapping.CZ = c1
>                 headers.put("topic","mytopic1");
>             }else if(body.contains("ORDER")){//后台下单日志
>                 //添加头信息
>                 //a1.sources.r1.selector.header = state
>                 //a1.sources.r1.selector.mapping.US = c1
>                 headers.put("topic","mytopic2");
>             }else{
>                 //a1.sources.r1.selector.default = c4
>                 //前两个判断没有进入则进入c4，flume.conf中不配置的话，则代表数据被抛弃掉
>             }
>     
>     
>             return event;
>         }
>     
>         @Override
>         public List<Event> intercept(List<Event> list) {
>             //清空list
>             addHeaderEventList.clear();
>             for (Event event : list) {
>                 addHeaderEventList.add(intercept(event));
>             }
>             return addHeaderEventList;
>         }
>     
>         @Override
>         public void close() {
>     
>         }
>         //静态内部类，用于返回拦截器实体
>         public static class Builder implements Interceptor.Builder{
>     
>     
>             @Override
>             public Interceptor build() {
>                 return new DistributeLogsInterceptor();
>             }
>     
>             @Override
>             public void configure(Context context) {
>     
>             }
>         }
>     }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMDQ5Nzcz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200122182706179.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIyMDQ5Nzcz,size_16,color_FFFFFF,t_70