flume+kafka+storm的集成使用

港控/mmm° 2022-06-18 11:25 215阅读 0赞

Flume-ng

Flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统。

Flume的文档可以看[http://flume.apache.org/FlumeUserGuide.html][http_flume.apache.org_FlumeUserGuide.html] 官方的英文文档 介绍的比较全面。

不过这里写写自己的见解

![Image 1][]![SouthEast][]

这个是flume的[架构][Link 1]图

从上图可以看到几个名词：

Agent: 一个Agent包含Source、Channel、Sink和其他的组件。Flume就是一个或多个Agent构成的。

Source：数据源。简单的说就是agent获取数据的入口 。

Channel：管道。数据流通和存储的通道。一个source必须至少和一个channel关联。

Sink：用来接收channel传输的数据并将之传送到指定的地方。传送成功后数据从channel中删除。

Flume具有高可扩展性 可随意组合：

![Image 1][]![SouthEast 1][]

注意 source是接收源 sink是发送源

![Image 1][]

![Image 1][]![SouthEast 2][]![SouthEast 3][]

上图是一个source将数据发给3个channel 其中的sink2将数据发给JMS ，sink3将数据发给另一个source。

总的来说flume的扩展性非常高 根据需要可随意组合。

现在在说说一个概念叫Event：

Event是flume的数据传输的基本单元。Flume本质上是将数据作为一个event从源头传到结尾。是由可选的Headers和载有数据的一个byte array构成。

代码结构：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1. /\*\*
2. \* Basic representation of a data object inFlume.
3. \* Provides access to data as it flows throughthe system.
4. \*/
5. public interface Event\{
6. /\*\*
7. \* Returns a map of name-valuepairs describing the data stored in the body.
8. \*/
9. public Map<String, String> getHeaders();
10. /\*\*
11. \* Set the event headers
12. \* @param headersMap of headers to replace the current headers.
13. \*/
14. public void setHeaders(Map<String, String> headers);
15. /\*\*
16. \* Returns the raw byte array of the datacontained in this event.
17. \*/
18. public byte\[\] getBody();
19. /\*\*
20. \* Sets the raw byte array of the datacontained in this event.
21. \* @param body Thedata.
22. \*/
23. public void setBody(byte\[\] body);
24. \}

![save_snippets.png][]

/**
 * Basic representation of a data object inFlume.
 * Provides access to data as it flows throughthe system.
 */
 public interface Event{
 /**
 * Returns a map of name-valuepairs describing the data stored in the body.
 */
 public Map<String, String> getHeaders();
 /**
 * Set the event headers
 * @param headersMap of headers to replace the current headers.
 */
 public void setHeaders(Map<String, String> headers);
 /**
 * Returns the raw byte array of the datacontained in this event.
 */
 public byte[] getBody();
 /**
 * Sets the raw byte array of the datacontained in this event.
 * @param body Thedata.
 */
 public void setBody(byte[] body);
 }

这个是网上找的flume channel ，source，sink的汇总

链接是[http://abloz.com/2013/02/26/flume-channel-source-sink-summary.html][http_abloz.com_2013_02_26_flume-channel-source-sink-summary.html]

<table> 
 <tbody> 
 <tr> 
 <td> 
 <table> 
 <tbody> 
 <tr> 
 <td style="background:rgb(221,221,221)"> Component </td> 
 <td style="background:rgb(221,221,221)"> Type </td> 
 <td style="background:rgb(221,221,221)"> Description </td> 
 <td style="background:rgb(221,221,221)"> Implementation Class </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> memory </td> 
 <td style="background:rgb(239,239,239)"> In-memory, fast, non-durable event transport </td> 
 <td style="background:rgb(239,239,239)"> MemoryChannel </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> file </td> 
 <td style="background:rgb(239,239,239)"> A channel for reading, writing, mapping, and manipulating a file </td> 
 <td style="background:rgb(239,239,239)"> FileChannel </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> jdbc </td> 
 <td style="background:rgb(239,239,239)"> JDBC-based, durable event transport (Derby-based) </td> 
 <td style="background:rgb(239,239,239)"> JDBCChannel </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> recoverablememory </td> 
 <td style="background:rgb(239,239,239)"> A durable channel implementation that uses the local file system for its storage </td> 
 <td style="background:rgb(239,239,239)"> RecoverableMemoryChannel </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.channel.PseudoTxnMemoryChannel </td> 
 <td style="background:rgb(239,239,239)"> Mainly for testing purposes. Not meant for production use. </td> 
 <td style="background:rgb(239,239,239)"> PseudoTxnMemoryChannel </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Channel </td> 
 <td style="background:rgb(239,239,239)"> (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own Channel impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> avro </td> 
 <td style="background:rgb(239,239,239)"> Avro Netty RPC event source </td> 
 <td style="background:rgb(239,239,239)"> AvroSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> exec </td> 
 <td style="background:rgb(239,239,239)"> Execute a long-lived Unix process and read from stdout </td> 
 <td style="background:rgb(239,239,239)"> ExecSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> netcat </td> 
 <td style="background:rgb(239,239,239)"> Netcat style TCP event source </td> 
 <td style="background:rgb(239,239,239)"> NetcatSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> seq </td> 
 <td style="background:rgb(239,239,239)"> Monotonically incrementing sequence generator event source </td> 
 <td style="background:rgb(239,239,239)"> SequenceGeneratorSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.source.StressSource </td> 
 <td style="background:rgb(239,239,239)"> Mainly for testing purposes. Not meant for production use. Serves as a continuous source of events where each event has the same payload. The payload consists of some number of bytes (specified bysize&nbsp;property, defaults to 500) where each byte has the signed value Byte.MAX_VALUE (0x7F, or 127). </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.source.StressSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> syslogtcp </td> 
 <td style="background:rgb(239,239,239)"> SyslogTcpSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> syslogudp </td> 
 <td style="background:rgb(239,239,239)"> SyslogUDPSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.source.avroLegacy.AvroLegacySource </td> 
 <td style="background:rgb(239,239,239)"> AvroLegacySource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.source.thriftLegacy.ThriftLegacySource </td> 
 <td style="background:rgb(239,239,239)"> ThriftLegacySource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.source.scribe.ScribeSource </td> 
 <td style="background:rgb(239,239,239)"> ScribeSource </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Source </td> 
 <td style="background:rgb(239,239,239)"> (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own Source impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> hdfs </td> 
 <td style="background:rgb(239,239,239)"> Writes all events received to HDFS (with support for rolling, bucketing, HDFS-200 append, and more) </td> 
 <td style="background:rgb(239,239,239)"> HDFSEventSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.<a title="Hbase知识库" href="http://lib.csdn.net/base/hbase" style="color:rgb(223,52,52); font-weight:bold" rel="nofollow">Hbase</a>.HBaseSink </td> 
 <td style="background:rgb(239,239,239)"> A simple sink that reads events from a channel and writes them to HBase. </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.HBaseSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.AsyncHBaseSink </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.AsyncHBaseSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> logger </td> 
 <td style="background:rgb(239,239,239)"> Log events at INFO level via configured logging subsystem (log4j by default) </td> 
 <td style="background:rgb(239,239,239)"> LoggerSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> avro </td> 
 <td style="background:rgb(239,239,239)"> Sink that invokes a pre-defined Avro protocol method for all events it receives (when paired with an avro source, forms tiered collection) </td> 
 <td style="background:rgb(239,239,239)"> AvroSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> file_roll </td> 
 <td style="background:rgb(239,239,239)"> RollingFileSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> irc </td> 
 <td style="background:rgb(239,239,239)"> IRCSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> null </td> 
 <td style="background:rgb(239,239,239)"> /dev/null for Flume – blackhole all events received </td> 
 <td style="background:rgb(239,239,239)"> NullSink </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Sink </td> 
 <td style="background:rgb(239,239,239)"> (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own Sink impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> ChannelSelector </td> 
 <td style="background:rgb(239,239,239)"> replicating </td> 
 <td style="background:rgb(239,239,239)"> ReplicatingChannelSelector </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> ChannelSelector </td> 
 <td style="background:rgb(239,239,239)"> multiplexing </td> 
 <td style="background:rgb(239,239,239)"> MultiplexingChannelSelector </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> ChannelSelector </td> 
 <td style="background:rgb(239,239,239)"> (custom type) </td> 
 <td style="background:rgb(239,239,239)"> Your own ChannelSelector impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> SinkProcessor </td> 
 <td style="background:rgb(239,239,239)"> default </td> 
 <td style="background:rgb(239,239,239)"> DefaultSinkProcessor </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> SinkProcessor </td> 
 <td style="background:rgb(239,239,239)"> failover </td> 
 <td style="background:rgb(239,239,239)"> FailoverSinkProcessor </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> SinkProcessor </td> 
 <td style="background:rgb(239,239,239)"> load_balance </td> 
 <td style="background:rgb(239,239,239)"> Provides the ability to load-balance flow over multiple sinks. </td> 
 <td style="background:rgb(239,239,239)"> LoadBalancingSinkProcessor </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> SinkProcessor </td> 
 <td style="background:rgb(239,239,239)"> (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own SinkProcessor impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Interceptor$Builder </td> 
 <td style="background:rgb(239,239,239)"> host </td> 
 <td style="background:rgb(239,239,239)"> HostInterceptor$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Interceptor$Builder </td> 
 <td style="background:rgb(239,239,239)"> timestamp </td> 
 <td style="background:rgb(239,239,239)"> TimestampInterceptor </td> 
 <td style="background:rgb(239,239,239)"> TimestampInterceptor$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Interceptor$Builder </td> 
 <td style="background:rgb(239,239,239)"> static </td> 
 <td style="background:rgb(239,239,239)"> StaticInterceptor$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Interceptor$Builder </td> 
 <td style="background:rgb(239,239,239)"> regex_filter </td> 
 <td style="background:rgb(239,239,239)"> RegexFilteringInterceptor$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> Interceptor$Builder </td> 
 <td style="background:rgb(239,239,239)"> (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own Interceptor$Builder impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer$Builder </td> 
 <td style="background:rgb(239,239,239)"> text </td> 
 <td style="background:rgb(239,239,239)"> BodyTextEventSerializer$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer$Builder </td> 
 <td style="background:rgb(239,239,239)"> avro_event </td> 
 <td style="background:rgb(239,239,239)"> FlumeEventAvroEventSerializer$Builder </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.SimpleHbaseEventSerializer </td> 
 <td style="background:rgb(239,239,239)"> SimpleHbaseEventSerializer </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.SimpleAsyncHbaseEventSerializer </td> 
 <td style="background:rgb(239,239,239)"> SimpleAsyncHbaseEventSerializer </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer </td> 
 <td style="background:rgb(239,239,239)"> org.apache.flume.sink.hbase.RegexHbaseEventSerializer </td> 
 <td style="background:rgb(239,239,239)"> RegexHbaseEventSerializer </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> HbaseEventSerializer </td> 
 <td style="background:rgb(239,239,239)"> Custom implementation of serializer for HBaseSink. (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own HbaseEventSerializer impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> AsyncHbaseEventSerializer </td> 
 <td style="background:rgb(239,239,239)"> Custom implementation of serializer for AsyncHbase sink. (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own AsyncHbaseEventSerializer impl. </td> 
 <td style="background:rgb(239,239,239)"> (custom FQCN) </td> 
 </tr> 
 <tr> 
 <td style="background:rgb(239,239,239)"> EventSerializer$Builder </td> 
 <td style="background:rgb(239,239,239)"> Custom implementation of serializer for all sinks except for HBaseSink and AsyncHBaseSink. (custom type as FQCN) </td> 
 <td style="background:rgb(239,239,239)"> Your own EventSerializer$Builder impl. </td> 
 </tr> 
 </tbody> 
 </table> &nbsp; </td> 
 </tr> 
 </tbody> 
</table>

下面介绍下kafka以及kafka和flume的整合

Kafka：

从这个链接抄了些内容下来[http://dongxicheng.org/search-engine/kafka/][http_dongxicheng.org_search-engine_kafka]

Kafka是Linkedin于2010年12月份开源的消息系统，它主要用于处理活跃的流式数据。活跃的流式数据在web网站应用中非常常见，这些数据包括网站的pv、用户访问了什么内容，搜索了什么内容等。 这些数据通常以日志的形式记录下来，然后每隔一段时间进行一次统计处理。

传统的日志分析系统提供了一种离线处理日志信息的可扩展方案，但若要进行实时处理，通常会有较大延迟。而现有的消（队列）系统能够很好的处理实时或者近似实时的应用，但未处理的数据通常不会写到磁盘上，这对于[Hadoop][]之类（一小时或者一天只处理一部分数据）的离线应用而言，可能存在问题。Kafka正是为了解决以上问题而设计的，它能够很好地离线和在线应用。

**2、 设计目标**

（1）数据在磁盘上存取代价为O(1)。一般数据在磁盘上是使用BTree存储的，存取代价为O（lgn）。

（2）高吞吐率。即使在普通的节点上每秒钟也能处理成百上千的message。

（3）显式分布式，即所有的producer、broker和consumer都会有多个，均为分布式的。

（4）支持数据并行加载到Hadoop中。

**3、 KafKa部署结构**

**![Image 1][]![SouthEast 4][]**

kafka是显式分布式架构，producer、broker（Kafka）和consumer都可以有多个。Kafka的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。几个基本概念：

（1）message（消息）是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。如果consumer订阅了这个主题，那么新发布的消息就会广播给这些consumer。

（2）Kafka是显式分布式的，多个producer、consumer和broker可以运行在一个大的集群上，作为一个逻辑整体对外提供服务。对于consumer，多个consumer可以组成一个group，这个message只能传输给某个group中的某一个consumer.

数据从producer推送到broker，接着consumer在从broker上拉取数据。Zookeeper是一个分布式服务框架 用来解决分布式应用中的数据管理问题等。

在kafka中 有几个重要概念producer生产者 consumer 消费者 topic 主题。

我们来实际开发一个简单的生产者消费者的例子。

生产者：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1. public classProducerTest \{
2. 
3. public static void main(String\[\] args) \{
4. Properties props = newProperties();
5. props.setProperty("metadata.broker.list","xx.xx.xx.xx:9092");
6. props.setProperty("serializer.class","kafka.serializer.StringEncoder");
7. props.put("request.required.acks","1");
8. ProducerConfigconfig = new ProducerConfig(props);
9. Producer<String, String> producer = newProducer<String, String>(config);
10. KeyedMessage<String, String> data = newKeyedMessage<String, String>("kafka","test-kafka");
11. try \{
12. producer.send(data);
13. \} catch (Exception e) \{
14. e.printStackTrace();
15. \}
16. producer.close();
17. \}
18. \}

![save_snippets.png][]

public classProducerTest {
 
 public static void main(String[] args) {
 Properties props = newProperties();
 props.setProperty("metadata.broker.list","xx.xx.xx.xx:9092");
 props.setProperty("serializer.class","kafka.serializer.StringEncoder");
 props.put("request.required.acks","1");
 ProducerConfigconfig = new ProducerConfig(props);
 Producer<String, String> producer = newProducer<String, String>(config);
 KeyedMessage<String, String> data = newKeyedMessage<String, String>("kafka","test-kafka");
 try {
 producer.send(data);
 } catch (Exception e) {
 e.printStackTrace();
 }
 producer.close(); 
 }
 }

上面的代码中的xx.xx.xx.xx是kafka server的地址.

上面代码的意思就是向主题 kafka中同步（不配置的话 默认是同步发射）发送了一个信息 是test-kafka.

下面来看看消费者：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1. public classConsumerTest extends Thread \{
2. private finalConsumerConnector consumer;
3. private final String topic;
4. 
5. public static voidmain(String\[\] args) \{
6. ConsumerTest consumerThread = newConsumerTest("kafka");
7. consumerThread.start();
8. \}
9. publicConsumerTest(String topic) \{
10. consumer =kafka.consumer.Consumer
11. .createJavaConsumerConnector(createConsumerConfig());
12. this.topic =topic;
13. \}
14. 
15. private staticConsumerConfig createConsumerConfig() \{
16. Properties props = newProperties();
17. props.put("zookeeper.connect","xx.xx.xx.xx:2181");
18. props.put("group.id", "0");
19. props.put("zookeeper.session.timeout.ms","10000");
20. // props.put("zookeeper.sync.time.ms", "200"); 
21. // props.put("auto.commit.interval.ms", "1000"); 
22. 
23. return newConsumerConfig(props);
24. 
25. \}
26. 
27. public void run()\{
28. 
29. Map<String,Integer> topickMap = new HashMap<String, Integer>();
30. topickMap.put(topic, 1);
31. Map<String, List<KafkaStream<byte\[\],byte\[\]>>> streamMap =consumer.createMessageStreams(topickMap);
32. KafkaStream<byte\[\],byte\[\]>stream = streamMap.get(topic).get(0);
33. ConsumerIterator<byte\[\],byte\[\]> it =stream.iterator();
34. System.out.println("--------------------------");
35. while(it.hasNext())\{
36. //
37. System.out.println("(consumer)--> " \+new String(it.next().message()));
38. \}
39. 
40. \}
41. \}

![save_snippets.png][]

public classConsumerTest extends Thread { 
 private finalConsumerConnector consumer; 
 private final String topic; 
 
 public static voidmain(String[] args) { 
 ConsumerTest consumerThread = newConsumerTest("kafka"); 
 consumerThread.start(); 
 } 
 publicConsumerTest(String topic) { 
 consumer =kafka.consumer.Consumer 
 .createJavaConsumerConnector(createConsumerConfig()); 
 this.topic =topic; 
 } 
 
 private staticConsumerConfig createConsumerConfig() { 
 Properties props = newProperties(); 
 props.put("zookeeper.connect","xx.xx.xx.xx:2181"); 
 props.put("group.id", "0"); 
 props.put("zookeeper.session.timeout.ms","10000"); 
 // props.put("zookeeper.sync.time.ms", "200"); 
 // props.put("auto.commit.interval.ms", "1000"); 
 
 return newConsumerConfig(props); 
 
 } 
 
 public void run(){ 
 
 Map<String,Integer> topickMap = new HashMap<String, Integer>();
 topickMap.put(topic, 1);
 Map<String, List<KafkaStream<byte[],byte[]>>> streamMap =consumer.createMessageStreams(topickMap);
 KafkaStream<byte[],byte[]>stream = streamMap.get(topic).get(0);
 ConsumerIterator<byte[],byte[]> it =stream.iterator();
 System.out.println("--------------------------");
 while(it.hasNext()){
 //
 System.out.println("(consumer)--> " +new String(it.next().message()));
 }
 
 } 
 }

上面的代码就是负责接收生产者发送过来的消息 [ 测试][Link 2]的时候先开启消费者 然后再运行生产者即可看到效果。

接下来 我们将flume 和kafka进行整合：

在flume的source数据源接收到数据后 通过管道 到达sink，我们需要写一个kafkaSink 来将sink从channel接收的数据作为kafka的生产者 将数据 发送给消费者。

具体代码：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1. public class KafkaSink extends AbstractSinkimplementsConfigurable \{
2. 
3. private static final Log logger = LogFactory.getLog(KafkaSink.class);
4. 
5. private Stringtopic;
6. private Producer<String, String>producer;
7. 
8. 
9. @Override
10. public Status process()throwsEventDeliveryException \{
11. 
12. Channel channel =getChannel();
13. Transaction tx =channel.getTransaction();
14. try \{
15. tx.begin();
16. Event e = channel.take();
17. if(e ==null) \{
18. tx.rollback();
19. return Status.BACKOFF;
20. \}
21. KeyedMessage<String,String> data = new KeyedMessage<String, String>(topic,newString(e.getBody()));
22. producer.send(data);
23. logger.info("Message: \{\}"\+new String( e.getBody()));
24. tx.commit();
25. return Status.READY;
26. \} catch(Exceptione) \{
27. logger.error("KafkaSinkException:\{\}",e);
28. tx.rollback();
29. return Status.BACKOFF;
30. \} finally \{
31. tx.close();
32. \}
33. \}
34. 
35. @Override
36. public void configure(Context context) \{
37. topic = "kafka";
38. Properties props = newProperties();
39. props.setProperty("metadata.broker.list","xx.xx.xx.xx:9092");
40. props.setProperty("serializer.class","kafka.serializer.StringEncoder");
41. // props.setProperty("producer.type", "async");
42. // props.setProperty("batch.num.messages", "1");
43. props.put("request.required.acks","1");
44. ProducerConfigconfig = new ProducerConfig(props);
45. producer = newProducer<String, String>(config);
46. \}
47. \}
48.

![save_snippets.png][]

public class KafkaSink extends AbstractSinkimplementsConfigurable {
 
 private static final Log logger = LogFactory.getLog(KafkaSink.class);
 
 private Stringtopic;
 private Producer<String, String>producer;
 
 
 @Override
 public Status process()throwsEventDeliveryException {
 
 Channel channel =getChannel();
 Transaction tx =channel.getTransaction();
 try {
 tx.begin();
 Event e = channel.take();
 if(e ==null) {
 tx.rollback();
 return Status.BACKOFF;
 }
 KeyedMessage<String,String> data = new KeyedMessage<String, String>(topic,newString(e.getBody()));
 producer.send(data);
 logger.info("Message: {}"+new String( e.getBody()));
 tx.commit();
 return Status.READY;
 } catch(Exceptione) {
 logger.error("KafkaSinkException:{}",e);
 tx.rollback();
 return Status.BACKOFF;
 } finally {
 tx.close();
 }
 }
 
 @Override
 public void configure(Context context) {
 topic = "kafka";
 Properties props = newProperties();
 props.setProperty("metadata.broker.list","xx.xx.xx.xx:9092");
 props.setProperty("serializer.class","kafka.serializer.StringEncoder");
 // props.setProperty("producer.type", "async");
 // props.setProperty("batch.num.messages", "1");
 props.put("request.required.acks","1");
 ProducerConfigconfig = new ProducerConfig(props);
 producer = newProducer<String, String>(config);
 }
 }

将此文件打成jar包 传到flume的lib下面 如果你也用的是maven的话 需要用到assembly 将依赖的jar包一起打包进去。

在flume的配置是如下：

**\[plain\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1.  agent1.sources = source1
2.  agent1.sinks = sink1
3.  agent1.channels =channel1
4.  
5.  \# Describe/configuresource1
6.  agent1.sources.source1.type= avro
7.  agent1.sources.source1.bind= localhost
8.  agent1.sources.source1.port= 44444
9.  \# Describe sink1
10. agent1.sinks.sink1.type= xx.xx.xx.KafkaSink(这是类的路径地址)
11. 
12. \# Use a channel whichbuffers events in memory
13. agent1.channels.channel1.type= memory
14. agent1.channels.channel1.capacity= 1000
15. agent1.channels.channel1.transactionCapactiy= 100
16. 
17. \# Bind the source andsink to the channel
18. agent1.sources.source1.channels= channel1
19. agent1.sinks.sink1.channel= channel1

![save_snippets.png][]

agent1.sources = source1
    agent1.sinks = sink1
    agent1.channels =channel1
     
    # Describe/configuresource1
    agent1.sources.source1.type= avro
    agent1.sources.source1.bind= localhost
    agent1.sources.source1.port= 44444
    # Describe sink1
    agent1.sinks.sink1.type= xx.xx.xx.KafkaSink(这是类的路径地址)
     
    # Use a channel whichbuffers events in memory
    agent1.channels.channel1.type= memory
    agent1.channels.channel1.capacity= 1000
    agent1.channels.channel1.transactionCapactiy= 100
     
    # Bind the source andsink to the channel
    agent1.sources.source1.channels= channel1
    agent1.sinks.sink1.channel= channel1

测试的话是avro的方式传送数据的 可以这样测试

bin/flume-ng avro-client--conf conf -H localhost -p 44444 -F/data/flumetmp/a

/data/flumetmp/a  这个为文件的地址.

测试的时候在本地 一定要把上面写的消费者程序打开 以便接收数据测试是否成功。

接下来我们介绍下storm然后将kafka的消费者和storm进行整合：

Storm：

Storm是一个分布式的实时消息处理系统。

Storm各个组件之间的关系：

![Image 1][]![SouthEast 5][]

Storm集群主要由一个主节点和一群工作节点（worker node）组成，通过 Zookeeper进行协调。

主节点：主节点通常运行一个后台程序 —— Nimbus，用于响应分布在集群中的节点，分配任务和监测故障。

工作节点：Supervisor,负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程。Nimbus和Supervisor之间的协调由zookeeper完成。

Worker：处理逻辑的进程，在其中运行着多个Task，每个task 是一组spout/blots的组合。

Topology:是storm的实时应用程序，从启动开始一直运行，只要有tuple过来 就会触发执行。拓扑：storm的消息流动很像一个拓扑结构。

**2.** stream是storm的核心概念，一个stream是一个持续的tuple序列，这些tuple被以分布式并行的方式创建和处理。

**3.** spouts是一个stream的源头，spouts负责从外部系统读取数据，并组装成tuple发射出去，tuple被发射后就开始再topology中传播。

**4.** bolt是storm中处理 数据的核心，storm中所有的数据处理都是在bolt中完成的

这里就简单介绍一些概念 具体的可以看些详细的教程。

我们接下来开始整合storm和kafka。

从上面的介绍得知storm的spout是负责从外部读取数据的 所以我们需要开发一个KafkaSpout 来作为kafka的消费者和storm的数据接收源。可以看看这个[https://github.com/HolmesNL/kafka-spout][https_github.com_HolmesNL_kafka-spout]。我在下面只写一个简单的可供测试。

具体代码：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1. public class KafkaSpout implements IRichSpout \{
2. 
3. private static final Log logger = LogFactory.getLog(KafkaSpout.class);
4. /\*\*
5. \*
6. \*/
7. private static final long serialVersionUID = -5569857211173547938L;
8. SpoutOutputCollector collector;
9. private ConsumerConnectorconsumer;
10. private Stringtopic;
11. 
12. public KafkaSpout(String topic) \{
13. this.topic = topic;
14. \}
15. 
16. @Override
17. public void open(Map conf, TopologyContext context,
18. SpoutOutputCollector collector) \{
19. this.collector = collector;
20. 
21. \}
22. 
23. private static ConsumerConfig createConsumerConfig() \{
24. Properties props = newProperties();
25. props.put("zookeeper.connect","xx.xx.xx.xx:2181");
26. props.put("group.id","0");
27. props.put("zookeeper.session.timeout.ms","10000");
28. //props.put("zookeeper.sync.time.ms", "200");
29. //props.put("auto.commit.interval.ms", "1000");
30. 
31. return new ConsumerConfig(props);
32. \}
33. 
34. @Override
35. public void close() \{
36. // TODOAuto-generated method stub
37. 
38. \}
39. 
40. @Override
41. public void activate() \{
42. this.consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());
43. Map<String, Integer> topickMap = newHashMap<String, Integer>();
44. topickMap.put(topic,new Integer(1));
45. Map<String, List<KafkaStream<byte\[\],byte\[\]>>>streamMap =consumer.createMessageStreams(topickMap);
46. KafkaStream<byte\[\],byte\[\]>stream = streamMap.get(topic).get(0);
47. ConsumerIterator<byte\[\],byte\[\]> it =stream.iterator();
48. while (it.hasNext()) \{
49. String value = newString(it.next().message());
50. System.out.println("(consumer)-->" \+ value);
51. collector.emit(new Values(value), value);
52. \}
53. 
54. \}
55. 
56. @Override
57. public void deactivate() \{
58. // TODOAuto-generated method stub
59. 
60. \}
61. 
62. private boolean isComplete;
63. 
64. @Override
65. public void nextTuple() \{
66. 
67. \}
68. 
69. @Override
70. public void ack(Object msgId) \{
71. // TODOAuto-generated method stub
72. 
73. \}
74. 
75. @Override
76. public void fail(Object msgId) \{
77. // TODOAuto-generated method stub
78. 
79. \}
80. 
81. @Override
82. public void declareOutputFields(OutputFieldsDeclarer declarer) \{
83. declarer.declare(new Fields("KafkaSpout"));
84. 
85. \}
86. 
87. @Override
88. public Map<String, Object> getComponentConfiguration() \{
89. // TODOAuto-generated method stub
90. return null;
91. \}
92. 
93. \}
94. 
95. public class FileBlots implementsIRichBolt\{
96. 
97. OutputCollector collector;
98. 
99. public void prepare(Map stormConf, TopologyContext context,
100. OutputCollector collector) \{
101. this.collector = collector;
102. 
103. \}
104. 
105. public void execute(Tuple input) \{
106. String line = input.getString(0);
107. for(String str : line.split("\\\\s+"))\{
108. List a = newArrayList();
109. a.add(input);
110. this.collector.emit(a,newValues(str));
111. \}
112. this.collector.ack(input);
113. \}
114. 
115. public void cleanup() \{
116. 
117. \}
118. 
119. public void declareOutputFields(OutputFieldsDeclarer declarer) \{
120. declarer.declare(new Fields("words"));
121. 
122. \}
123. 
124. public Map<String, Object> getComponentConfiguration() \{
125. // TODOAuto-generated method stub
126. return null;
127. \}
128. 
129. \}
130. public class WordsCounterBlots implementsIRichBolt\{
131. 
132. OutputCollector collector;
133. Map<String, Integer> counter;
134. 
135. public void prepare(Map stormConf, TopologyContext context,
136. OutputCollector collector) \{
137. this.collector = collector;
138. this.counter =new HashMap<String, Integer>();
139. 
140. \}
141. 
142. public void execute(Tuple input) \{
143. String word = input.getString(0);
144. Integer integer = this.counter.get(word);
145. if(integer !=null)\{
146. integer +=1;
147. this.counter.put(word, integer);
148. \}else\{
149. this.counter.put(word, 1);
150. \}
151. System.out.println("execute");
152. Jedis jedis = JedisUtils.getJedis();
153. jedis.incrBy(word, 1);
154. System.out.println("=============================================");
155. this.collector.ack(input);
156. \}
157. 
158. public void cleanup() \{
159. for(Entry<String, Integer> entry :this.counter.entrySet())\{
160. System.out.println("------:"\+entry.getKey()+"=="\+entry.getValue());
161. \}
162. 
163. \}
164. 
165. public void declareOutputFields(OutputFieldsDeclarer declarer) \{
166. 
167. 
168. \}
169. 
170. public Map<String, Object> getComponentConfiguration() \{
171. // TODOAuto-generated method stub
172. return null;
173. \}
174. 
175. \}

![save_snippets.png][]

public class KafkaSpout implements IRichSpout {
 
 private static final Log logger = LogFactory.getLog(KafkaSpout.class);
 /**
 *
 */
 private static final long serialVersionUID = -5569857211173547938L;
 SpoutOutputCollector collector;
 private ConsumerConnectorconsumer;
 private Stringtopic;
 
 public KafkaSpout(String topic) {
 this.topic = topic;
 }
 
 @Override
 public void open(Map conf, TopologyContext context,
 SpoutOutputCollector collector) {
 this.collector = collector;
 
 }
 
 private static ConsumerConfig createConsumerConfig() {
 Properties props = newProperties();
 props.put("zookeeper.connect","xx.xx.xx.xx:2181");
 props.put("group.id","0");
 props.put("zookeeper.session.timeout.ms","10000");
 //props.put("zookeeper.sync.time.ms", "200");
 //props.put("auto.commit.interval.ms", "1000");
 
 return new ConsumerConfig(props);
 }
 
 @Override
 public void close() {
 // TODOAuto-generated method stub
 
 }
 
 @Override
 public void activate() {
 this.consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());
 Map<String, Integer> topickMap = newHashMap<String, Integer>();
 topickMap.put(topic,new Integer(1));
 Map<String, List<KafkaStream<byte[],byte[]>>>streamMap =consumer.createMessageStreams(topickMap);
 KafkaStream<byte[],byte[]>stream = streamMap.get(topic).get(0);
 ConsumerIterator<byte[],byte[]> it =stream.iterator();
 while (it.hasNext()) {
 String value = newString(it.next().message());
 System.out.println("(consumer)-->" + value);
 collector.emit(new Values(value), value);
 }
 
 }
 
 @Override
 public void deactivate() {
 // TODOAuto-generated method stub
 
 }
 
 private boolean isComplete;
 
 @Override
 public void nextTuple() {
 
 }
 
 @Override
 public void ack(Object msgId) {
 // TODOAuto-generated method stub
 
 }
 
 @Override
 public void fail(Object msgId) {
 // TODOAuto-generated method stub
 
 }
 
 @Override
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
 declarer.declare(new Fields("KafkaSpout"));
 
 }
 
 @Override
 public Map<String, Object> getComponentConfiguration() {
 // TODOAuto-generated method stub
 return null;
 }
 
 }
 
 public class FileBlots implementsIRichBolt{
 
 OutputCollector collector;
 
 public void prepare(Map stormConf, TopologyContext context,
 OutputCollector collector) {
 this.collector = collector;
 
 }
 
 public void execute(Tuple input) {
 String line = input.getString(0);
 for(String str : line.split("\\s+")){
 List a = newArrayList();
 a.add(input); 
 this.collector.emit(a,newValues(str));
 }
 this.collector.ack(input);
 }
 
 public void cleanup() {
 
 }
 
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
 declarer.declare(new Fields("words"));
 
 }
 
 public Map<String, Object> getComponentConfiguration() {
 // TODOAuto-generated method stub
 return null;
 }
 
 }
 public class WordsCounterBlots implementsIRichBolt{
 
 OutputCollector collector;
 Map<String, Integer> counter;
 
 public void prepare(Map stormConf, TopologyContext context,
 OutputCollector collector) {
 this.collector = collector;
 this.counter =new HashMap<String, Integer>();
 
 }
 
 public void execute(Tuple input) {
 String word = input.getString(0);
 Integer integer = this.counter.get(word);
 if(integer !=null){
 integer +=1;
 this.counter.put(word, integer);
 }else{
 this.counter.put(word, 1);
 }
 System.out.println("execute");
 Jedis jedis = JedisUtils.getJedis();
 jedis.incrBy(word, 1);
 System.out.println("=============================================");
 this.collector.ack(input);
 }
 
 public void cleanup() {
 for(Entry<String, Integer> entry :this.counter.entrySet()){
 System.out.println("------:"+entry.getKey()+"=="+entry.getValue());
 }
 
 }
 
 public void declareOutputFields(OutputFieldsDeclarer declarer) {
 
 
 }
 
 public Map<String, Object> getComponentConfiguration() {
 // TODOAuto-generated method stub
 return null;
 }
 
 }

Topology测试：

**\[java\]** [ view plain][view plain] [ copy][view plain] [ print][view plain] [?][view plain]

1.  public class KafkaTopology \{
2.  
3.  public static void main(String\[\] args) \{
4.  try \{
5.  JedisUtils.initialPool("xx.xx.xx.xx", 6379);
6.  \} catch (Exception e) \{
7.  e.printStackTrace();
8.  \}
9.  
10. TopologyBuilder builder = newTopologyBuilder(); builder.setSpout("kafka",new KafkaSpout("kafka"));
11. builder.setBolt("file-blots",new FileBlots()).shuffleGrouping("kafka");
12. builder.setBolt("words-counter",new WordsCounterBlots(),2).fieldsGrouping("file-blots",new Fields("words"));
13. Config config = new Config();
14. config.setDebug(true);
15. LocalCluster local = newLocalCluster();
16. local.submitTopology("counter", config, builder.createTopology());
17. \}
18. \}

![save_snippets.png][]

public class KafkaTopology {
     
          public static void main(String[] args) {
               try {
                     JedisUtils.initialPool("xx.xx.xx.xx", 6379);
               } catch (Exception e) {
                     e.printStackTrace();
               }
              
               TopologyBuilder builder = newTopologyBuilder();           builder.setSpout("kafka",new KafkaSpout("kafka"));
               builder.setBolt("file-blots",new FileBlots()).shuffleGrouping("kafka");
               builder.setBolt("words-counter",new WordsCounterBlots(),2).fieldsGrouping("file-blots",new Fields("words"));
               Config config = new Config();
               config.setDebug(true);
                     LocalCluster local = newLocalCluster();
                     local.submitTopology("counter", config, builder.createTopology());
          }
    }

至此flume + kafka+storm的整合就写完了。注意 这个是 初始学习阶段做的测试 不可正式用于线上环境，在写本文之时 已经离测试过去了一段时间 所以可能会有些错误 请见谅

[http_flume.apache.org_FlumeUserGuide.html]: http://flume.apache.org/FlumeUserGuide.html
[Image 1]: 
[SouthEast]: /images/20220618/c3b6a759ce344f4cbfd55e79b14b0da8.png
[Link 1]: http://lib.csdn.net/base/architecture
[SouthEast 1]: /images/20220618/df14c9f212db4ab196c538711a5eba3d.png
[SouthEast 2]: https://img-blog.csdn.net/20140121114003328?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhjdmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast
[SouthEast 3]: /images/20220618/65dc9f4a41aa46a9a26143d66ef5c117.png
[view plain]: http://blog.csdn.net/zxcvg/article/details/18600335/#
[save_snippets.png]: /images/20220618/7484a62a4ba14560a59c416c8ffcd31e.png
[http_abloz.com_2013_02_26_flume-channel-source-sink-summary.html]: http://abloz.com/2013/02/26/flume-channel-source-sink-summary.html
[http_dongxicheng.org_search-engine_kafka]: http://dongxicheng.org/search-engine/kafka/
[Hadoop]: http://lib.csdn.net/base/hadoop
[SouthEast 4]: /images/20220618/2cc73fb566ba48b9b132ddaa5c7ba390.png
[Link 2]: http://lib.csdn.net/base/softwaretest
[SouthEast 5]: /images/20220618/2f04dced8c2c4071b1b7b33616513d05.png
[https_github.com_HolmesNL_kafka-spout]: https://github.com/HolmesNL/kafka-spout