channel选择器副本机制
我们现在的需求就是实时的监听一个文件的变化，然后将变化的数据上传到HDFS，另外需要在本地文件系统保存副本，所以此次的sink去向有两个，一个是HDFS，另外一个就是本地文件系统。
其实我们可以用一个flume就可以实现将第一个flume的sink分别设置为两个去向，为什么下图要多用两个flume呢？是因为有时HDFSsink端写数据较慢，生产数据较快，那就有可能生产大于消费，刚开始还可以在channel进行缓存，如果channel也满了就会崩掉，多两个flume也是为了增加缓冲。
上图我们可以看到第一个flume从source监听数据，然后交给两个channel，这就需要两个channel收到的数据是一致的，所以channel选择器需要配置成副本，然后两个channel分别发往对应的sink，这里sink采用的是avro，它是用于两个flume之间联系的组件，如果两个flume需要联系，上一个flume的sink就要配置avro，下端的flume的source也要配置相应的avro。
avro的sink就是将数据发往指定IP的某个端口，avro的source就是从某个IP地址的指定端口进行获取数据。
Flume2配置的就是avrosource，然后sink端就是HDFS。
Flume3同样配置avro，但是sink端配置的是file_roll，用于传输到本地文件系统
Flume1
a1.sources = r1 a1.sinks = k1 k2 a1.channels = c1 c2 a1.sources.r1.selector.type=replicating a1.sources.r1.type = taildir a1.sources.r1.channels = c1 a1.sources.r1.positionFile = /opt/module/flume/position/position.json a1.sources.r1.filegroups = f1 a1.sources.r1.filegroups.f1 = /home/hadoop/data3/data a1.sinks.k1.type = avro a1.sinks.k1.hostname=hadoop102 a1.sinks.k1.port=4141 a1.sinks.k2.type = avro a1.sinks.k2.hostname=hadoop102 a1.sinks.k2.port=4142 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 a1.sources.r1.channels = c1 c2 a1.sinks.k1.channel = c1 a1.sinks.k2.channel = c2
Flume2
a2.sources = r1 a2.sinks = k1 a2.channels = c1 a2.sources.r1.type = avro a2.sources.r1.bind=hadoop102 a2.sources.r1.port=4141 a2.sinks.k1.type = hdfs # 配置hdfs存储的路径，这里我设置的是动态路径，按照时间命名的文件夹 a2.sinks.k1.hdfs.path=hdfs://hadoop102:9000/flume/%Y%m%d/%H # 给存储到HDFS的文件加个前缀标明该数据为日志信息 a2.sinks.k1.hdfs.filePrefix=logs- # 是否按照时间进行滚动文件夹，对应上面配置的动态文件夹 a2.sinks.k1.hdfs.round=true a2.sinks.k1.hdfs.roundValue=1 # 设置时间量级为秒、分钟还是小时 a2.sinks.k1.hdfs.roundUnit=hour # 是否使用本地时间戳，我测试了一下，如果这里设置成false，HDFS目录不会发生变化，可能是hdfs配置的动态时间路径就是按照时间戳配置的 a2.sinks.k1.hdfs.useLocalTimeStamp=true # 积攒多少个Event才会flush到HDFS a2.sinks.k1.hdfs.batchSize=1000 # 文件类型 a2.sinks.k1.hdfs.fileType=DataStream # 多久会生成一个新的文件，如果监听的文件没有变化，尽管到了时间也不会产生新文件，产生新文件需要事件去触发 a2.sinks.k1.hdfs.rollInterval=30 # 每个文件大小达到多少会进行滚动 a2.sinks.k1.hdfs.rollSize=134217700 # 设置文件的滚动与事件无关 a2.sinks.k1.hdfs.rollCount=0 a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 100 a2.sources.r1.channels = c1 a2.sinks.k1.channel = c1
Flume3
a3.sources = r1 a3.sinks = k1 a3.channels = c1 a3.sources.r1.type = avro a3.sources.r1.bind=hadoop102 a3.sources.r1.port=4142 a3.sinks.k1.type = file_roll a3.sinks.k1.sink.directory = /home/hadoop/data4/flume a3.channels.c1.type = memory a3.channels.c1.capacity = 1000 a3.channels.c1.transactionCapacity = 100 a3.sources.r1.channels = c1 a3.sinks.k1.channel = c1
这样当我们向data文件中追加数据，整个flume流程就将我们追加的数据以副本机制分别传输到HDFS和本地文件系统。
