kafka整合flume实例

素颜马尾好姑娘i 2023-01-18 15:20 22阅读 0赞

###  ###

*  一 环境准备
 *   *  1.1 kafka集群部署
     *  1.2 flume下载
     *  1.3 解压flume安装包
     *  1.4 flume环境变量配置
     *  1.5 flume配置文件
 *  二 创建java工程
 *  三 测试
 *   *  3.1启动 zookeeper集群
     *  3.2 启动 kafka集群
     *  3.3 启动 flume
     *  3.4 创建topic
     *  3.5 创建消费者
     *  3.6 运行java代码

# 一 环境准备 #

## 1.1 kafka集群部署 ##

kafka集群搭建详见：[kafka集群搭建][kafka]。本文不再进行讲解。

## 1.2 flume下载 ##

flume下载地址：archive.apache.org/dist/flume

## 1.3 解压flume安装包 ##

在 node0上解压flume压缩包

tar -xvf apache-flume-1.6.0-bin.tar.gz  -C /opt/app/

## 1.4 flume环境变量配置 ##

解压后，在配置文件 /etc/profile添加

export FLUME_HOME=/opt/app/apache-flume-1.6.0-bin                       
     export PATH=$PATH:$FLUME_HOME/bin  
     source /etc/profile

## 1.5 flume配置文件 ##

cd /opt/app/apache-flume-1.6.0/conf
    cp flume-env.sh.template flume-env.sh
    vi flume-env.sh

修改java环境变量

export JAVA_HOME=/usr/local/java/jdk1.8.0_221

![在这里插入图片描述][20210507003443377.png]

cp  flume-conf.properties.template  flume-conf.properties 
    vim  flume-conf.properties

flume-conf.properties 添加如下内容

# Name the components on this agent -
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = avro
    a1.sources.r1.bind = node0
    a1.sources.r1.port = 41414
    
    
    # Describe the sink
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.topic = testflume
    #a1.sinks.k1.topic = test1
    a1.sinks.k1.brokerList = node0:9092,node1:9092,node2:9092
    a1.sinks.k1.requiredAcks = 1
    a1.sinks.k1.batchSize = 20
    
    
    # Use a channel which buffers events in memory
    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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70]

# 二 创建java工程 #

创建一个java工程，引入flume相关jar包，主类RpcClientDemo 代码如下

public class RpcClientDemo { 
            
            public static void main(String[] args) throws InterruptedException { 
                    MyRpcClientFacade client = new MyRpcClientFacade();
                    // Initialize client with the remote Flume agent's host and port
                    client.init("node0", 41414);
    
                    // Send 10 events to the remote Flume agent. That agent should be
                    // configured to listen with an AvroSource.
                    for (int i = 0; i < 11; i++) { 
                            String sampleData = "Hello Flume!ERROR" + i;
                            if (i % 2 == 0) { 
                                    sampleData = "Hello Flume!WARNNING" + i;
                            }
                            client.sendDataToFlume(sampleData);
                            System.out.println("发送数据：" + sampleData);
                            Thread.sleep(200);
                    }
    
                    client.cleanUp();
            }
    }
    
    class MyRpcClientFacade { 
            private RpcClient client;
            private String hostname;
            private int port;
    
            public void init(String hostname, int port) { 
                    // Setup the RPC connection
                    this.hostname = hostname;
                    this.port = port;
                    this.client = RpcClientFactory.getDefaultInstance(hostname, port);
                    // Use the following method to create a thrift client (instead of the
                    // above line):
                    // this.client = RpcClientFactory.getThriftInstance(hostname, port);
            }
    
            public void sendDataToFlume(String data) { 
                    // 使用flume的event封装消息，该消息需要指定编码方式
                    // data就是要发送的数据，字符串
                    Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
    
                    // 发送数据
                    try { 
                            client.append(event);
                    } catch (EventDeliveryException e) { 
                            // clean up and recreate the client
                            client.close();
                            client = null;
                            client = RpcClientFactory.getDefaultInstance(hostname, port);
                            // Use the following method to create a thrift client (instead of
                            // the above line):
                            // this.client = RpcClientFactory.getThriftInstance(hostname, port);
                    }
            }
    
            public void cleanUp() { 
                    // 断开 RPC 连接
                    client.close();
            }
    }

# 三 测试 #

## 3.1启动 zookeeper集群 ##

在node1 、node2，node3上执行

zkServer.sh start

## 3.2 启动 kafka集群 ##

在 node0 、node1、node上启动kafka集群  
进入kafka安装目录

kafka-server-start.sh config/server.properties

## 3.3 启动 flume ##

在node0上 进入flume 安装目录，执行

bin/flume-ng agent -n a1 -c conf -f conf/flume-conf.properties -Dflume.root.logger=DEBUG,console

## 3.4 创建topic ##

重新打开node0，创建topic testflume

kafka-topics.sh  --create  --partitions 3 --replication-factor 2 --topic testflume --zookeeper node1:2181,node2:2181
      kafka-topics.sh  --zookeeper node1  --list
       kafka-topics.sh  --zookeeper node1  --describe --topic testflume

![在这里插入图片描述][20210507004516857.png]  
![在这里插入图片描述][20210507004522130.png]  
![在这里插入图片描述][20210507004554658.png]

## 3.5 创建消费者 ##

在 node0上创建消费者

kafka-console-consumer.sh   --zookeeper node1:2181 --from begining  --topic testflume

## 3.6 运行java代码 ##

运行java代码 发送信息到flume 由kafka 消费  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 1]  
然后可以看到kafka消费者消费了客户端  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 2]

[kafka]: https://blog.csdn.net/whandgdh/article/details/116279215
[20210507003443377.png]: /images/20221021/19dee24198b642aa85f3029f7317ff78.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70]: /images/20221021/3914418c049b418a93e734d5798b8143.png
[20210507004516857.png]: /images/20221021/cbdaea69f9a14a8f89ab9e4ac22eea59.png
[20210507004522130.png]: /images/20221021/a8e1a195686f4191b5889d345ee80b7e.png
[20210507004554658.png]: /images/20221021/bcc67d45437d408087596a7d0dd43c94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 1]: /images/20221021/b192c5ca46fd4c47829e2fc309667761.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 2]: /images/20221021/6d575144492749148c732e0542de9274.png