分享

Flume

 IT技术武馆 2014-08-05
在使用Flume的时候,请确保你电脑里面已经搭建好Hadoop、Hbase、Zookeeper以及Flume。本文将以最新版的Hadoop-2.2.0、Hbase-0.96.0、Zookeeper-3.4.5以及Flume-1.4.0为例进行说明。
 1、本程序一共用了三台集群搭建集群,这三台机器的Hostname分别为master、node1、node2;master机器是Hadoop以及Hbase集群的master。三台机器上分别启动的进程如下:

  1. [wyp@master ~]$ jps
  2. 2973 HRegionServer
  3. 4083 Jps
  4. 2145 DataNode
  5. 3496 HMaster
  6. 2275 NodeManager
  7. 1740 NameNode
  8. 2790 QuorumPeerMain
  9. 1895 ResourceManager

  10. [wyp@node1 ~]$ jps
  11. 7801 QuorumPeerMain
  12. 11669 DataNode
  13. 29419 Jps
  14. 11782 NodeManager
  15. 29092 HRegionServer

  16. [wyp@node2 ~]$ jps
  17. 2310 DataNode
  18. 2726 HRegionServer
  19. 2622 QuorumPeerMain
  20. 3104 Jps
  21. 2437 NodeManager
复制代码

2、以master机器作为flume数据的源、并将数据发送给node1机器上的flume,最后node1机器上的flume将数据插入到Hbase中。master机器上的flume和node1机器上的flume中分别做如下的配置:
在master的$FLUME_HOME/conf/目录下创建以下文件(文件名随便取),并做如下配置,这是数据的发送端:


  1. [wyp@master conf]$ vim example.conf
  2. agent.sources = baksrc
  3. agent.channels = memoryChannel
  4. agent.sinks = remotesink

  5. agent.sources.baksrc.type = exec
  6. agent.sources.baksrc.command = tail -F /home/wyp/Documents/data/data.txt
  7. agent.sources.baksrc.checkperiodic = 1000

  8. agent.channels.memoryChannel.type = memory
  9. agent.channels.memoryChannel.keep-alive = 30
  10. agent.channels.memoryChannel.capacity = 10000
  11. agent.channels.memoryChannel.transactionCapacity = 10000

  12. agent.sinks.remotesink.type = avro
  13. agent.sinks.remotesink.hostname = node1
  14. agent.sinks.remotesink.port = 23004
  15. agent.sinks.remotesink.channel = memoryChannel
复制代码

在node1的$FLUME_HOME/conf/目录下创建以下文件(文件名随便取),并做如下配置,这是数据的接收端:

  1. [wyp@node1 conf]$ vim example.conf
  2. agent.sources = avrosrc
  3. agent.channels = memoryChannel
  4. agent.sinks = fileSink

  5. agent.sources.avrosrc.type = avro
  6. agent.sources.avrosrc.bind = node1
  7. agent.sources.avrosrc.port = 23004
  8. agent.sources.avrosrc.channels = memoryChannel

  9. agent.channels.memoryChannel.type = memory
  10. agent.channels.memoryChannel.keep-alive = 30
  11. agent.channels.memoryChannel.capacity = 10000
  12. agent.channels.memoryChannel.transactionCapacity =10000

  13. agent.sinks.fileSink.type = hbase
  14. agent.sinks.fileSink.table = wyp
  15. agent.sinks.fileSink.columnFamily = cf
  16. agent.sinks.fileSink.column = charges
  17. agent.sinks.fileSink.serializer =
  18.                   org.apache.flume.sink.hbase.RegexHbaseEventSerializer
  19. agent.sinks.fileSink.channel = memoryChannel
复制代码

这两个文件配置的含义我就不介绍了,自己google一下吧。

3、在master机器和node1机器上分别启动flume服务进程:
  1. [wyp@master apache-flume-1.4.0-bin]$ bin/flume-ng agent
  2.                                   --conf conf
  3.                                   --conf-file conf/example.conf
  4.                                   --name agent
  5.                                   -Dflume.root.logger=INFO,console

  6. [wyp@node1 apache-flume-1.4.0-bin]$ bin/flume-ng agent
  7.                                   --conf conf
  8.                                   --conf-file conf/example.conf
  9.                                   --name agent
  10.                                   -Dflume.root.logger=INFO,console
复制代码

当分别在node1和master机器上启动上面的进程之后,在node1机器上将会输出以下的信息:
  1. 2014-01-20 22:41:56,179 (pool-3-thread-1)
  2. [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.
  3.                                       handleUpstream(NettyServer.java:171)]
  4. [id: 0x16c775c5, /192.168.142.161:42201 => /192.168.142.162:23004] OPEN
  5. 2014-01-20 22:41:56,182 (pool-4-thread-1)
  6. [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.
  7.                                       handleUpstream(NettyServer.java:171)]
  8. [id: 0x16c775c5, /192.168.142.161:42201 => /192.168.142.162:23004]
  9.                                       BOUND: /192.168.142.162:23004
  10. 2014-01-20 22:41:56,182 (pool-4-thread-1)
  11. [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.
  12.                                       handleUpstream(NettyServer.java:171)]
  13. [id: 0x16c775c5, /192.168.142.161:42201 => /192.168.142.162:23004]
  14.                                       CONNECTED: /192.168.142.161:42201
复制代码

在master机器上将会输出以下的信息:
  1. 2014-01-20 22:42:16,625 (lifecycleSupervisor-1-0)
  2. [INFO - org.apache.flume.sink.AbstractRpcSink.
  3. createConnection(AbstractRpcSink.java:205)]
  4. Rpc sink remotesink: Building RpcClient with hostname: node1, port: 23004
  5. 2014-01-20 22:42:16,625 (lifecycleSupervisor-1-0)
  6. [INFO - org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:126)]
  7. Attempting to create Avro Rpc client.
  8. 2014-01-20 22:42:19,639 (lifecycleSupervisor-1-0)
  9. [INFO - org.apache.flume.sink.AbstractRpcSink.start(AbstractRpcSink.java:300)]
  10. Rpc sink remotesink started.
复制代码

这样暗示node1上的flume和master上的flume已经连接成功了。

  4、如何测试?可以写一个脚本往/home/wyp/Documents/data/data.txt(见上面master机器上flume上面的配置)文件中追加东西:

  1. for i in {1..1000000}; do
  2.     echo "test flume to Hbase $i" >>
  3.            /home/wyp/Documents/data/data.txt;
  4.     sleep 0.1;
  5. done
复制代码

 运行上面的脚本,这样将每隔0.1秒往/home/wyp/Documents/data/data.txt文件中添加内容,这样master上的flume将会接收到/home/wyp/Documents/data/data.txt文件内容的变化,并变化的内容发送到node1机器上的flume,node1机器上的flume把接收到的内容插入到Hbase的wyp表中的cf:charges列中(见上面的配置)。

出现问题解决办法:
本文是以最新版的Flume和最新办的Hbase进行整合,在整合的过程中将会出现flume依赖包版本问题,解决方法是用
$HADOOP_HOME/share/hadoop/common/lib/guava-11.0.2.jar替换$FLUME_HOME/lib/guava-10.0.1.jar包;
用$HADOOP_HOME/share/hadoop/common/lib/protobuf-java-2.5.0.jar替换$HBASE_HOME/lib/protobuf-java-2.4.0.jar包。然后再启动步骤三的两个进程。

    本站是提供个人知识管理的网络存储空间,所有内容均由用户发布,不代表本站观点。请注意甄别内容中的联系方式、诱导购买等信息,谨防诈骗。如发现有害或侵权内容,请点击一键举报。
    转藏 分享 献花(0

    0条评论

    发表

    请遵守用户 评论公约

    类似文章 更多