Flume Source 实例

Flume Source 实例

  1. 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
  2. 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
  3. 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文件夹内放入文件。

  4. 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
  5. 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
  6. 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服务

  7. 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都附带着模式信息
上一篇:Redis结合Lua脚本实现高并发原子性操作


下一篇:Bug预防体系(上千bug分析后总结的最佳实践)