Flume个人学习总结

WHAT

一个分布式、可靠、高可用的海量日志采集系统

• 可靠性：保证数据不丢失
• 可扩展性：各组件数目可扩展
• 高性能：吞吐率很高，能满足海量数据收集需求
• 可管理性：可动态增加和删除组件

整体架构

Flume的组件：

• Event：Flume数据传输的基本单元
• Client： 是一个将原始log包装成Event并发送它们到一个或多个Agent的实体
• Agent：一个Agent包含source、channel、sink和其他组件，利用这些组件将events从一个节点传输到另一个节点或最终目的
  • Source：负责接收event，或通过特殊机制产生event，并将其批量的放入一个或多个channel source必须至少和一个channel关联！
  • Channel：位于source和sink之间，用于缓存event
  • Sink：负责将event传输到下一跳或者最终目的，成功完成后将event从channel中删除

---

WHY

Hadoop提供了一个中央化的存储系统，有利于集中式的数据分析和数据共享。但数据是分散在各个离散的设备上，数据保存在传统的存储设备和系统中。对于这些非结构数据（日志）的收集，主要用到Flume。而结构化数据收集（传统数据库与Hadoop同步）可以通过Sqoop全量导入。而后续，可以通过DataBus或Camus增量导入。

HOW

• 说明：四个参数分别为source（avro）方式，0.0.0.0为本地模式，端口设置为41414，最终路径为 logAgent.sinks.hdfsSink.hdfs.path = hdfs://192.168.99.130:9000/flume/record3/%Y-%m-%d/%H%M 的临时文件

Paste_Image.png

• 去flume的bin目录下运行以下语句

[xc2046@master apache-flume-1.6.0-bin]$ bin/flume-ng agent --conf /home/xc2046/apache-flume-1.6.0-bin/conf --conf-file /home/xc2046/apache-flume-1.6.0-bin/conf/avro.properties --name logAgent -Dflume.root.logger=DEBUG,console -Dflume.monitoring.type=http -Dflume.monitoring.port=34545

• 通过一段java代码，链接设置的Flume，向HDFS上的文件写入数据

package com.example.flume;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import java.nio.charset.Charset;

public class FlumeExample {

    public static void main(String[] args){
        MyRpcClientFacade clientFacade = new MyRpcClientFacade();
        clientFacade.init("192.168.99.130", 41414);
        //Send 10 event to the remote Flume agent.That agent should be configured to listen with AvroSource
        String msg = "Hello Flume!";
        for(int i=0;i<10;i++){
            clientFacade.sendDataToFlume(msg);
            System.out.println("sended");
        }
        clientFacade.cleanUp();
    }

}

class MyRpcClientFacade{

    private RpcClient client;

    private String hostname;

    private int port;

    public void init(String hostname,int port){
        this.hostname = hostname;
        this.port = port;
        this.client = RpcClientFactory.getDefaultInstance(hostname,port);
        //this.client = RpcClientFactory.getThriftInstance(hostname,port);
    }

    public void sendDataToFlume(String data){
        //create a Flume Event Object that encapsulates the data
        Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
        //send the event
        try {
            client.append(event);
        } catch (EventDeliveryException e) {
            client.close();
            client = null;
            client = RpcClientFactory.getDefaultInstance(hostname,port);
            // Thrift Client
            //client = RpcClientFactory.getThriftInstance(hostname,port);
            e.printStackTrace();
        }
    }

    public void cleanUp(){
        //Close the RPC connection
        client.close();
    }

}

• 去HDFS上cat刚刚Flume生成的临时文件

Paste_Image.png

---

只是最基础的Flume一些介绍、操作，希望以后能有更多的理解。推荐一下心_的方向作者的Flume浅析