Flume与Kafka对接「建议收藏」

Flume与Kafka对接「建议收藏」引言flume为什么要与kafka对接?我们都知道flume可以跨节点进行数据的传输,那么flume与sparkstreaming对接不好吗?主要是flume对接到kafka的topic,可以给多个consumergroup去生成多条业务线。虽然flume中的channelselector中的副本策略也可以做多给多个sink传输数据,但是每个channelselector都是很消耗资源的。文章目录一、flume采集的数据发往一个topic二、flume采集的数据发往多个topic总结.

大家好,又见面了,我是你们的朋友全栈君。

引言
flume为什么要与kafka对接?
我们都知道flume可以跨节点进行数据的传输,那么flume与spark streaming对接不好吗?主要是flume对接到kafka的topic,可以给多个consumer group去生成多条业务线。虽然flume中的channel selector中的副本策略也可以给多个sink传输数据,但是每个channel selector都是很消耗资源的。其次,kafka也可以起到一个消峰的作用


一、flume采集的数据发往一个topic

这里为了方便测试,我采用的是netcat source、memory channel、kafka sink,当然你也可以采用你自己想要的方式配置flume,只需要根据官方文档修改对应的source和channel即可。

necat-flume-kafka.conf的配置文件如下:

#Name
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#Source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

#Channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#Sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = wjt
a1.sinks.k1.brokerList = node01:9092,node02:9092,node03:9092
a1.sinks.k1.kafka.flumeBatchSize = 20 
a1.sinks.k1.kafka.producer.acks = 1 
a1.sinks.k1.kafka.producer.linger.ms = 1 

#Bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

其中你只需要修改sink中的topic和brokerList即可,当然你也可以增加其他的配置
1、启动kafka消费者
在这里插入图片描述
2、启动flume
在这里插入图片描述
3、启动netcat的客户端并发送几条数据
在这里插入图片描述
4、观察到kafka consumer很快就消费到了数据
在这里插入图片描述

二、flume采集的数据发往多个topic

如果数据有多种类型,比如点赞数据、评论数据、喜欢数据等等,是不是就要发往不同的topic去分析数据,这时候就需要用到flume的拦截器来做分类。
flume可以给event加上头信息,结合channel selector来发往不同的sink。
在flume官方文档可以看到:
在这里插入图片描述
意思是:如果你的event的头信息(k-v类型)包含一个topic字段,那么这个event将会被发送到对应的topic,并覆盖你配置的kafka.topic

拦截器的代码:

package wjt.demo;

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;

/** * @description: * @author: wanjintao * @time: 2020/8/29 11:45 */
public class myInterceptor 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中是否有“Hello”来决定是否添加头信息
        if (body.contains("hello")) { 
   

            //4. 有hello添加“wan”头信息
            headers.put("topic", "www1");

        } else { 
   

            //4. 没有hello添加“tao”头信息
            headers.put("topic", "www2");

        }

        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 myInterceptor();
        }

        @Override
        public void configure(Context context) { 
   

        }
    }

}

你只需要修改单个事件拦截的代码即可,我这里是如果数据包含hello,将会给事件加上header(topic,www1),反之则给事件加上header(topic,www2),打包上传至flume/lib目录下

netcat-flume-typekafka.conf的配置文件:

#Name
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#Source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

#Interceptor
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = wjt.demo.myInterceptor$Builder

#Channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#Sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = wjt
a1.sinks.k1.brokerList = node01:9092,node02:9092,node03:9092
a1.sinks.k1.kafka.flumeBatchSize = 20 
a1.sinks.k1.kafka.producer.acks = 1 
a1.sinks.k1.kafka.producer.linger.ms = 1 

#Bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

你只需要将a1.sources.r1.interceptors.i1.type的值改为你上面的拦截器的全类名$Builder即可
1、先启动consumer1和consumer2(flume启动顺序都是先启动服务端在启动客户端)
在这里插入图片描述
在这里插入图片描述
2、启动flume
在这里插入图片描述
3、启动netcat客户端
在这里插入图片描述
4、观察consumer消费的topic可以看到,www1只接受到了包含hello的数据,www2只接受到了没有包含hello的数据
在这里插入图片描述
在这里插入图片描述

总结

很多时候flume官方文档可以帮助我们解决很多自己想要的业务场景,我们要更多地去查看官方文档

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/152405.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • cnpm安装步骤[通俗易懂]

    cnpm安装步骤[通俗易懂]安装nodeJS官网下载:https://nodejs.org/zh-cn/download/releases/选版本点击下载然后下载后缀名为msi,因为安装简单二、创建文件夹安装完成后我们打开它的目录创建两个文件夹(后面配置环境变量需要)node_cachenode_global三、配置npm的全局模块的存放路径、cache的路径win+r输入cmd打开命令提示符窗口,输入:npmconfigsetprefix”选择刚刚创建node_global文件路径”np

    2022年10月16日
  • Sqoop问题解决:运行报错java.lang.RuntimeException: Could not load db driver class: com.mysql.jdbc.Driver

    Sqoop问题解决:运行报错java.lang.RuntimeException: Could not load db driver class: com.mysql.jdbc.DriverSqoop问题解决:运行报错报错信息:java.lang.RuntimeException:Couldnotloaddbdriverclass:com.mysql.jdbc.Driver原因分析:未将mysql关系型数据库驱动包放到sqoop/lib目录下解决方法:将mysql关系型数据库驱动包放到sqoop/lib目录下这里需要下载mysql关系型数据库驱动包放到本地/opt/software/下mysql依赖包下载链接:https://pan.baidu.com/s

  • goland激活3月最新在线激活

    goland激活3月最新在线激活,https://javaforall.cn/100143.html。详细ieda激活码不妨到全栈程序员必看教程网一起来了解一下吧!

  • 大数据运维面试题整理

    大数据运维面试题整理HDFS总结:1、HDFS是如何解决大规模数据的存储和管理的2、HDFS的架构原理和各核心组件的作用及关系3、HDFS文件系统数据的读写流程4、HDFS的HA的架构原理及核心5、HDFS的Federation机制6、HDSF的HA和Federation的区别…

  • Pycharm连接并调用服务器「建议收藏」

    Pycharm连接并调用服务器「建议收藏」Pycharm可以与服务器建立连接,把相应的项目同步到服务器上,并且可以通过Pycharm直接使用服务器的解释器运行相应程序,实现Pycharm编程,服务器运行的效果。具体步骤如下:1.建立一个服务器连接Pycharm的“Tools”-》“Deployment”-》“Configuration”2.创建一个SFTP3.为该项目添加一个SSH解释器。因为前面已经添加好了服务器连接,所以这里直接选择已经设置好的就可以,如果没有已经设置好的,可以重新添加。配置好SSH之后,选择Next,设置本地项目

  • springboot 如何优雅的抛出异常

    springboot 如何优雅的抛出异常

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号