Flume Source 实例

hongblue

　　Flume Source 实例

• 　　Avro Source
  　　监听avro端口，接收外部avro客户端数据流。跟前面的agent的Avro Sink可以组成多层拓扑结构。
  
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=avro a1.sources.s1.bind=vm1 a1.sources.s1.port=41414     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方法：
  
  

  1	flume-ng avro-client --host vm1 --port 41414 --filename a.log

  　　
  
• 　　Exec Source
  　　启动source的时候执行unix命令，并期望不断获取数据，比如tail -f命令。
  　　Exec Source有可能会丢失数据，可以考虑使用Spooling Dircetory Source。
  
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=exec a1.sources.s1.command= tail -f /home/flume/a.log     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　执行命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方法：
  　　向unix命令监听的文件追加内容，
  
  

  1	echo 'hello, everyone!' >> a.log

  
  
• 　　Spooling Directory Source
  　　监控某个文件夹，将新产生的文件解析成event，解析方式是可插拔的，默认是LINE，将新文件中的每行数据转换成一个event。文件解析完成后，该文件名字被追加.completed。
  　　Spooling Dircetory Source相对于Exec Source更可靠，它不会丢失数据（甚至被重启或杀死）。为了这种特性，只能是不可修改的、名字不重复的文件才可使用这个source：
  　　1）监控文件夹内的文件被flume使用后不能进行修改。
  　　2）文件名字不能重复，如果已经存在了xx.completed，不能在放入名字叫xx的文件。否则报错，停止。
  　　3）被监控文件夹里的子文件夹不处理
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=spooldir a1.sources.s1.spoolDir=/home/flume/a     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方法：
  　　向/home/flume/a文件夹内放入文件。
  
  
• 　　Netcat Source
  　　监听端口数据。类似于nc -k -l [host] [port]命令.
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=netcat a1.sources.s1.bind=localhost a1.sources.s1.port=41414     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方式：
  
  

  1	telnet localhost 41414

  
  
• 　　Sequence Generator Source
  　　不断生成从0开始的数字，主要用于测试
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=seq     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  
  
• 　　Syslog Source
  　　读取syslog数据，它分为：Syslog TCP Source、Multiport Syslog TCP Source、Syslog UDP Source
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=syslogtcp a1.sources.s1.host=localhost a1.sources.s1.port=41414     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方法：
  　　需要syslog服务
  
  
• 　　Http Source
  　　通过Http获取event，可以是GET、POST方式，GET方式应该在测试时使用。一个HTTP Request被handler解析成若干个event，这组event在一个事务里。
  　　如果handler抛异常，http状态是400；如果channel满了，http状态是503。
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

  a1.sources=s1 a1.sinks=k1 a1.channels=c1     a1.sources.s1.channels=c1 a1.sinks.k1.channel=c1     a1.sources.s1.type=http a1.sources.s1.port=41414     a1.sinks.k1.type=logger     a1.channels.c1.type=memory a1.channels.c1.capacity=1000 a1.channels.c1.transactionCapacity=100

  　　启动命令：
  
  

  1	flume-ng agent --conf conf/ --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

  　　测试方法：
  
  

  1	curl -XPOST http://vm1:41414 -d '[{"headers" : {"timestamp" : "434324343","host" : "random_host.example.com"},"body" : "random_body"},{"headers" : {"namenode" : "namenode.example.com","datanode" : "random_datanode.example.com"},"body" : "really_random_body"}]'

  　　参数的格式是：
  
  

  1 2 3 4 5 6 7 8 9 10 11 12 13

  [{     "headers" : {          "timestamp" : "434324343",          "host" : "random_host.example.com"     },     "body" : "random_body" }, {     "headers" : {          "namenode" : "namenode.example.com",          "datanode" : "random_datanode.example.com"      },     "body" : "really_random_body" }]

  　　Http Source Handler：
  

  Handler类型	说明
  JSONHandler	默认值，将文本输入的每行转换成一个event
  BlobHandler	读取avro文件，将其中的每条avro记录转换成一个event，每个event都附带着模式信息

  　　
  
  

　　

　　

　　



来自为知笔记(Wiz)