flume-agent实例

flume
    多种适配，多样化的数据收集
    核心概念
        event：一条消息
        client：访问者
        agent：
            重要组件Sources、Channels、Sinks。Interspactor、Selecter
        
        
kafka
    吞吐量大，高并发场景下使用

注意：flume的agent配置文件不允许有空格。

    
    一、flume打印内容到控制台
        1、创建一个agent(使用avroSource接收网络流在flume的控制台打印)配置文件agent1.conf
            cd /usr/local/flume/
            vi /conf/agent1.conf
                agent1.sources=as1
                agent1.channels=c1
                agent1.sinks=s1

                agent1.sources.as1.type=avro
                agent1.sources.as1.bind=0.0.0.0            ##接收任意ip发送的数据
                agent1.sources.as1.port=21111            ##在21111端口上监听
                agent1.sources.as1.channels=c1
                agent1.channels.c1.type=memory

                agent1.sinks.s1.type=logger
                agent1.sinks.s1.channel=c1
        2、启动agent1(每30秒检查agent1.conf文件一次，检查该文件是否有变化，有变化则马上生效)，将输出打印在控制台上
            bin/flume-ng agent --conf conf/ -Dflume.root.logger=DEBUG,console -n agent1 -f conf/agent1.conf
        3、使用java代码生产log4j日志输出到flume
            
        3、验证agent，一种是flume控制台测试，一种是java代码通过log4j写日志
            1)bin/flume-ng avro-client --conf conf/ -H localhost -p 21111 -F ~/a        ##将~目录下的a文件内容写入到flume
            2)使用java类将log4j的日志写入到flume的agent中
                log4j.properties配置文件
                    log4j.rootLogger=INFO,flume
                    log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
                    log4j.appender.flume.Hostname = 192.168.1.33                                    ##flume启动agent所在的节点ip
                    log4j.appender.flume.Port = 21111                                                ##flume启动agent监听的端口号
                    log4j.appender.flume.UnsafeMode = true
                
                java代码    
                    public class FlumeProducer {
                        public static void main(String[] args) throws Exception {
                            final Logger logger = Logger.getLogger(FlumeProducer.class);
                            while (true) {
                                logger.info("logger datetime :" + System.currentTimeMillis());
                                Thread.sleep(1000);
                            }
                        }
                    }

    二、flume生成avroLog文件写入到hdfs中，存放到不同的/IP/日期/文件夹中
        1、创建一个agent(使用avroSource接收网络流写入到hdfs)配置文件agent2.conf
            cd /usr/local/flume/
            vi /conf/agent2.conf
                agent2.sources=source1
                agent2.channels=channel1
                agent2.sinks=sink1

                agent2.sources.source1.type=avro
                agent2.sources.source1.bind=0.0.0.0
                agent2.sources.source1.port=44444
                agent2.sources.source1.channels=channel1
                
                agent2.sources.source1.interceptors = i1 i2
                agent2.sources.source1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Builder
                agent2.sources.source1.interceptors.i1.preserveExisting = true
                agent2.sources.source1.interceptors.i1.useIP = true
                agent2.sources.source1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

                
                agent2.channels.channel1.type=memory
                agent2.channels.channel1.capacity=10000
                agent2.channels.channel1.transactionCapacity=1000
                agent2.channels.channel1.keep-alive=30

                agent2.sinks.sink1.type=hdfs
                agent2.sinks.sink1.channel=channel1
                agent2.sinks.sink1.hdfs.path=hdfs://ns1/flume/events/%{host}/%Y-%m-%d            ##flume将文件写入到hdfs的路径
                agent2.sinks.sink1.hdfs.filePrefix=avroLog-                                        ##flume生成文件的前缀
                agent2.sinks.sink1.hdfs.fileSuffix=.log                                            ##flume生成文件的后缀
                agent2.sinks.sink1.hdfs.fileType=DataStream                                        ##flume生成文件的类型，DataStream或SequenceFile
                agent2.sinks.sink1.hdfs.writeFormat=Text
                agent2.sinks.sink1.hdfs.rollInterval=0
                agent2.sinks.sink1.hdfs.rollSize=10000
                agent2.sinks.sink1.hdfs.rollCount=0
                agent2.sinks.sink1.hdfs.idleTimeout=5
        2、启动agent2(每30秒检查agent1.conf文件一次，检查该文件是否有变化，有变化则马上生效)，将内容写入到hdfs的/flume/events/中
            bin/flume-ng agent --conf conf/ -Dflume.monitoring.type=http -Dflume.monitoring.port=34343 -n agent2 -f conf/agent2.conf
        3、使用java代码生产log4j日志输出到flume
            log4j.properties配置文件
                log4j.rootLogger=INFO,flume
                log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
                log4j.appender.flume.Hostname = 192.168.1.33                                    ##flume启动agent所在的节点ip
                log4j.appender.flume.Port = 21111                                                ##flume启动agent监听的端口号
                log4j.appender.flume.UnsafeMode = true
                
            java代码    
                public class FlumeProducer {
                    public static void main(String[] args) throws Exception {
                        final Logger logger = Logger.getLogger(FlumeProducer.class);
                        while (true) {
                            logger.info("logger datetime :" + System.currentTimeMillis());
                            Thread.sleep(1000);
                        }
                    }
                }            
        4、验证agent2是否成功写入到hdfs的/flume/events/文件夹下
            hdfs dfs -ls -h -R /flume/events/IP/yyyy-MM-dd/                                        ##如果存在一个或多个avroLog.timestamp.log文件表示成功
            
　　三、使用Socket客户端写入到flume中，flume保存文件到本地

　　　　1、创建agent_tcp.conf（接收socket客户端发送的数据然后写入到Linux本地）

　　　　　　cd /usr/local/flume

　　　　　　vi conf/agent_tcp.conf

　　　　　　　　agent_tcp.sources=as1
　　　　　　　　agent_tcp.channels=c1
　　　　　　　　agent_tcp.sinks=s1

　　　　　　　　agent_tcp.sources.as1.type=syslogtcp
　　　　　　　　agent_tcp.sources.as1.bind=0.0.0.0
　　　　　　　　agent_tcp.sources.as1.port=21111
　　　　　　　　agent_tcp.sources.as1.channels=c1

　　　　　　　　agent_tcp.channels.c1.type=memory
　　　　　　　　agent_tcp.channels.c1.capacity=10000
　　　　　　　　agent_tcp.channels.c1.transactionCapacity=10000
　　　　　　　　agent_tcp.channels.c1.keep-alive=120
　　　　　　　　agent_tcp.channels.c1.byteCapacityBufferPercentage=20
　　　　　　　　agent_tcp.channels.c1.byteCapacity=800000
　　
　　　　　　　　agent_tcp.sinks.s1.type=file_roll
　　　　　　　　agent_tcp.sinks.s1.rollSize=10000
　　　　　　　　agent_tcp.sinks.s1.sink.directory =/home/lefuBigDataDev/clouds/flume/logs
　　　　　　　　agent_tcp.sinks.s1.channel=c1
　　　　2、启动flume的agent_tcp.conf

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

　　　　3、java代码socket客户端

　　　　　　package com.left.clouds.cluster.flume.test;

            　　import java.io.InputStream;
            　　import java.io.OutputStream;
            　　import java.net.Socket;

　　            import org.junit.Before;
      　　      import org.junit.Test;

      　　      public class TestFlume {

      　　          private Socket client = null;
      　　          InputStream in = null;
      　　          OutputStream out = null;
                
      　　          @Before
      　　          public void before(){
      　　              try {
      　　                  client = new Socket("192.168.0.218", 21111);
      　　              } catch (Exception e) {
      　　                  e.printStackTrace();
      　　              }
      　　          }
                
      　　          @Test
      　　          public void sender() {
      　　              try {
      　　                  out = client.getOutputStream();
      　　                  int i = 0;
      　　                  while(true){
      　　                      out.write(("device-"+(i++)+(" ")).getBytes());
      　　                      Thread.sleep(4000);
      　　                      System.out.println("第："+i+"次发送...");
      　　                  }
      　　              } catch (Exception e) {
      　　                  e.printStackTrace();
      　　              }　　              
      　　          }

      　　      }

 

Flume-1.6.0中包含了kafka的source，agent配置文件实例如下
front_agent_kafka.sources=as1
front_agent_kafka.channels=c1
front_agent_kafka.sinks=s1

front_agent_kafka.sources.as1.type=org.apache.flume.source.kafka.KafkaSource
front_agent_kafka.sources.as1.zookeeperConnect=192.168.0.20:2181
front_agent_kafka.sources.as1.topic=test
front_agent_kafka.sources.as1.groupId=flume
front_agent_kafka.sources.as1.batchSize=100
front_agent_kafka.sources.as1.channels=c1
                              
front_agent_kafka.channels.c1.type=memory
front_agent_kafka.channels.c1.capacity=10000
front_agent_kafka.channels.c1.transactionCapacity=10000
front_agent_kafka.channels.c1.keep-alive=120
front_agent_kafka.channels.c1.byteCapacityBufferPercentage=20
front_agent_kafka.channels.c1.byteCapacity=800000

front_agent_kafka.sinks.s1.type=com.lefukj.flume.sinks.JdbcSink
front_agent_kafka.sinks.s1.channel=c1