flume之常用sink详解

偏执的太偏执、 2021-06-24 15:57 614阅读 0赞

**1、Logger Sink **  
  
记录INFO级别的日志，一般用于调试。前面介绍Source时候用到的Sink都是这个类型的Sink  
  
必须配置的属性：      
            type    logger  
            maxBytesToLog    16    Maximum number of bytes of the Event body to log  
  
  
注：要求必须在 --conf 参数指定的目录下有 log4j的配置文件，可以通过-Dflume.root.logger=INFO,console在命令启动时手动指定log4j参数

案例：前面的例子都是这种类型的Sink

**2、File Roll Sink**  
  
在本地文件系统中存储事件。每隔指定时长生成文件保存这段时间内收集到的日志信息。  
  
属性说明：  
            type    file\_roll  
            sink.directory    必填，文件被存储的目录  
            sink.rollInterval    30    滚动文件每隔30秒（应该是每隔30秒钟单独切割数据到一个文件的意思）。如果设置为0，则禁止滚动，从而导致所有数据被写入到一个文件。  
            sink.serializer    TEXT    Other possible options include avro\_event or the FQCN of an implementation of EventSerializer.Builder interface.  
            batchSize    100

实例：

a1.sources  =  r1
    a1.sinks  =  k1
    a1.channels  =  c1
    
    a1.sources.r1.type  = http
    a1.sources.r1.port  = 6666
    a1.sources.r1.channels  =  c1
    
    a1.channels.c1.type  =  memory
    a1.channels.c1.capacity  =  1000
    a1.channels.c1.transactionCapacity  =  100
    
    a1.sinks.k1.type  = file_roll
    a1.sinks.k1.sink.directory = /home/park/work/apache-flume-1.6.0-bin/mysink
    a1.sinks.k1.channel  =  c1

3、Avro Sink  
  
非常重要，是实现多级流动 和 扇出流(1到多) 扇入流(多到1) 的基础。 flume使用avro rpc实现多个flume节点进行连接。  
  
必要属性说明:    
            type    avro  
            hostname    必填，he hostname or IP address to bind to.  
            port    必填，The port \# to listen on.

实例1：扇入，多个节点a1。。。an 流入到b1

1）a1...an的配置：

a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    a1.sources.r1.type=http
    a1.sources.r1.port=8888
    a1.sources.r1.channels=c1
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=1000
    
    a1.sinks.k1.type=avro
    a1.sinks.k1.hostname=192.168.242.138
    a1.sinks.k1.port=9988
    a1.sinks.k1.channel=c1

2）b1的配置：

b1.sources=r1
    b1.sinks=k1
    b1.channels=c1
    
    b1.sources.r1.type=avro
    b1.sources.r1.bind=0.0.0.0
    b1.sources.r1.port=9988
    b1.sources.r1.channels=c1
    
    b1.channels.c1.type=memory
    b1.channels.c1.capacity=1000
    b1.channels.c1.transactionCapacity=1000
    
    b1.sinks.k1.type=logger
    b1.sinks.k1.channel=c1

启动flume后，会发现数据汇总到了b1服务器上。

实例2：扇出操作，1各节点a1流出到多个节点b1...bn

1）a1节点配置：

a1.sources=r1
    a1.sinks=k1 k2
    a1.channels=c1 c2
    
    a1.sources.r1.type=http
    a1.sources.r1.port=8888
    a1.sources.r1.channels=c1 c2
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=1000
    a1.channels.c2.type=memory
    a1.channels.c2.capacity=1000
    a1.channels.c2.transactionCapacity=1000
    
    a1.sinks.k1.type=avro
    a1.sinks.k1.hostname=192.168.242.138
    a1.sinks.k1.port=9988
    a1.sinks.k1.channel=c1 
    a1.sinks.k2.type=avro
    a1.sinks.k2.hostname=192.168.242.135
    a1.sinks.k2.port=9988
    a1.sinks.k2.channel=c2

**注：如果要实现event数据流的扇出，需要配置多个channel和sink，如果只配置一个channel、多个sink，那么event数据通过channel后会被多个sink互斥的消费掉。**

2）b1...bn配置：

b1.sources=r1
    b1.sinks=k1
    b1.channels=c1
    
    #描述/配置Source
    b1.sources.r1.type=avro
    b1.sources.r1.bind=0.0.0.0
    b1.sources.r1.port=9988
    b1.sources.r1.channels=c1
    
    b1.channels.c1.type=memory
    b1.channels.c1.capacity=1000
    b1.channels.c1.transactionCapacity=1000
    
    #描述Sink
    b1.sinks.k1.type=logger
    b1.sinks.k1.channel=c1

启动flume后，event数据通过a1后会分成两股流通过channel1和channel2，然后分发到b1...bn

**4、HDFS Sink**  
  
此Sink将事件写入到Hadoop分布式文件系统HDFS中。前它支持创建文本文件和序列化文件，对这两种格式都支持压缩。 这些文件可以分卷，按照指定的时间或数据量或事件的数量为基础，它还通过类似时间戳或机器属性对数据进行 buckets/partitions 操作。HDFS的目录路径可以包含将要由HDFS替换格式的转移序列用以生成存储事件的目录/文件名。  
**注：**使用这个Sink要求hadoop必须已经安装好，以便Flume可以通过hadoop提供的jar包与HDFS进行通信，此版本hadoop必须支持sync()调用。  
  
必要属性说明:  
                type   HDFS  
                hdfs.path    必填，HDFS 目录路径 (eg hdfs://namenode/flume/webdata/)  
                hdfs.filePrefix    FlumeData    Flume在目录下创建文件的名称前缀  
                hdfs.fileSuffix    –    追加到文件的名称后缀 (eg .avro - 注: 日期时间不会自动添加)  
                hdfs.inUsePrefix    –    Flume正在处理的文件所加的前缀  
                hdfs.inUseSuffix    .tmp    Flume正在处理的文件所加的后缀  
  
实例：

a1.sources=r1
    a1.sinks=k1
    a1.channels=c1
    
    a1.sources.r1.type=http
    a1.sources.r1.port=8888
    a1.sources.r1.channels=c1
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=1000
    
    a1.sinks.k1.type=hdfs
    a1.sinks.k1.hdfs.path=hdfs://0.0.0.0:9000/ppp
    a1.sinks.k1.channel=c1

参考：http://www.cnblogs.com/itdyb/p/6270893.html