Flume：单数据源多输出 之 选择器实现

需求

使用Flume-1监控文件Hive日志，Flume-1将变动内容传递给Flume-2，Flume-2负责存储到HDFS。同时Flume-1将变动内容传递给Flume-3，Flume-3负责输出到Local FileSystem。

实现

第一步：在flume目录下创建文件夹job，然后在其下面创建配置文件：

• flume-file-flume.conf

  # Name the components on this agent
  a1.sources = r1
  a1.sinks = k1 k2
  a1.channels = c1 c2
  # 将数据流复制给所有channel
  a1.sources.r1.selector.type = replicating
  
  # Describe/configure the source
  a1.sources.r1.type = exec
  a1.sources.r1.command = tail -F /usr/local/apache-hive-3.1.1/logs/hive.log
  a1.sources.r1.shell = /bin/bash -c
  
  # Describe the sink
  a1.sinks.k1.type = avro
  a1.sinks.k1.hostname = hcmaster 
  a1.sinks.k1.port = 4141
  
  a1.sinks.k2.type = avro
  a1.sinks.k2.hostname = hcmaster
  a1.sinks.k2.port = 4142
  
  # Describe the channel
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 1000
  a1.channels.c1.transactionCapacity = 100
  
  a1.channels.c2.type = memory
  a1.channels.c2.capacity = 1000
  a1.channels.c2.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a1.sources.r1.channels = c1 c2
  a1.sinks.k1.channel = c1
  a1.sinks.k2.channel = c2
  

• flume-flume-dir.conf

  # Name the components on this agent
  a3.sources = r1
  a3.sinks = k1
  a3.channels = c2
  
  # Describe/configure the source
  a3.sources.r1.type = avro
  a3.sources.r1.bind = hcmaster
  a3.sources.r1.port = 4142
  
  # Describe the sink
  a3.sinks.k1.type = file_roll
  a3.sinks.k1.sink.directory = /usr/local/test
  
  # Describe the channel
  a3.channels.c2.type = memory
  a3.channels.c2.capacity = 1000
  a3.channels.c2.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a3.sources.r1.channels = c2
  a3.sinks.k1.channel = c2
  

• flume-flume-hdfs.conf

  # Name the components on this agent
  a2.sources = r1
  a2.sinks = k1
  a2.channels = c1
  
  # Describe/configure the source
  a2.sources.r1.type = avro
  a2.sources.r1.bind = hcmaster
  a2.sources.r1.port = 4141
  
  # Describe the sink
  a2.sinks.k1.type = hdfs
  a2.sinks.k1.hdfs.path = hdfs://hcmaster:8020/flume2/%Y%m%d/%H
  #上传文件的前缀
  a2.sinks.k1.hdfs.filePrefix = flume2-
  #是否按照时间滚动文件夹
  a2.sinks.k1.hdfs.round = true
  #多少时间单位创建一个新的文件夹
  a2.sinks.k1.hdfs.roundValue = 1
  #重新定义时间单位
  a2.sinks.k1.hdfs.roundUnit = hour
  #是否使用本地时间戳
  a2.sinks.k1.hdfs.useLocalTimeStamp = true
  #积攒多少个Event才flush到HDFS一次
  a2.sinks.k1.hdfs.batchSize = 100
  #设置文件类型，可支持压缩
  a2.sinks.k1.hdfs.fileType = DataStream
  #多久生成一个新的文件
  a2.sinks.k1.hdfs.rollInterval = 600
  #设置每个文件的滚动大小大概是128M
  a2.sinks.k1.hdfs.rollSize = 134217700
  #文件的滚动与Event数量无关
  a2.sinks.k1.hdfs.rollCount = 0
  #最小冗余数
  a2.sinks.k1.hdfs.minBlockReplicas = 1
  
  # Describe the channel
  a2.channels.c1.type = memory
  a2.channels.c1.capacity = 1000
  a2.channels.c1.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a2.sources.r1.channels = c1
  a2.sinks.k1.channel = c1
  

第二步：启动flume 第三步：启动hive 在hive shell中做一些简单操作

第四步：查看结果：