Kafka+Storm+HDFS整合实践
在基于Hadoop平台的很多应用场景中，我们需要对数据进行离线和实时分析，离线分析可以很容易地借助于Hive来实现统计分析，但是对于实时的需求Hive就不合适了。实时应用场景可以使用Storm，它是一个实时处理系统，它为实时处理类应用提供了一个计算模型，可以很容易地进行编程处理。为了统一离线和实时计算，一般情况下，我们都希望将离线和实时计算的数据源的集合统一起来作为输入，然后将数据的流向分别经由实时系统和离线分析系统，分别进行分析处理，这时我们可以考虑将数据源（如使用Flume收集日志）直接连接一个消息中间件，如Kafka，可以整合Flume+Kafka，Flume作为消息的Producer，生产的消息数据（日志数据、业务请求数据等等）发布到Kafka中，然后通过订阅的方式，使用Storm的Topology作为消息的Consumer，在Storm集群中分别进行如下两个需求场景的处理：
- 直接使用Storm的Topology对数据进行实时分析处理
- 整合Storm+HDFS，将消息处理后写入HDFS进行离线分析处理
实时处理，只要开发满足业务需要的Topology即可，不做过多说明。这里，我们主要从安装配置Kafka、Storm，以及整合Kafka+Storm、整合Storm+HDFS、整合Kafka+Storm+HDFS这几点来配置实践，满足上面提出的一些需求。配置实践使用的软件包如下所示：
- zookeeper-3.4.5.tar.gz
- kafka_2.9.2-0.8.1.1.tgz
- apache-storm-0.9.2-incubating.tar.gz
- hadoop-2.2.0.tar.gz
程序配置运行所基于的操作系统为CentOS 5.11。
Kafka安装配置
我们使用3台机器搭建Kafka集群：
|
1
2
3
|
192.168.4.142 h1192.168.4.143 h2192.168.4.144 h3 |
在安装Kafka集群之前，这里没有使用Kafka自带的Zookeeper，而是独立安装了一个Zookeeper集群，也是使用这3台机器，保证Zookeeper集群正常运行。
首先，在h1上准备Kafka安装文件，执行如下命令：
|
1
2
3
4
5
|
cd /usr/local/wget http://mirror.bit.edu.cn/apache/kafka/0.8.1.1/kafka_2.9.2-0.8.1.1.tgztar xvzf kafka_2.9.2-0.8.1.1.tgzln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafkachown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka |
修改配置文件/usr/local/kafka/config/server.properties，修改如下内容：
|
1
2
|
broker.id=0zookeeper.connect=h1:2181,h2:2181,h3:2181/kafka |
这里需要说明的是，默认Kafka会使用ZooKeeper默认的/路径，这样有关Kafka的ZooKeeper配置就会散落在根路径下面，如果你有其他的应用也在使用ZooKeeper集群，查看ZooKeeper中数据可能会不直观，所以强烈建议指定一个chroot路径，直接在zookeeper.connect配置项中指定：
|
1
|
zookeeper.connect=h1:2181,h2:2181,h3:2181/kafka |
而且，需要手动在ZooKeeper中创建路径/kafka，使用如下命令连接到任意一台ZooKeeper服务器：
|
1
2
|
cd /usr/local/zookeeperbin/zkCli.sh |
在ZooKeeper执行如下命令创建chroot路径：
|
1
|
create /kafka '' |
这样，每次连接Kafka集群的时候（使用--zookeeper选项），也必须使用带chroot路径的连接字符串，后面会看到。
然后，将配置好的安装文件同步到其他的h2、h3节点上：
|
1
2
|
scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h2:/usr/local/scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h3:/usr/local/ |
最后，在h2、h3节点上配置，执行如下命令：
|
1
2
3
|
cd /usr/local/ln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafkachown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka |
并修改配置文件/usr/local/kafka/config/server.properties内容如下所示：
|
1
2
3
|
broker.id=1 # 在h1修改broker.id=2 # 在h2修改 |
因为Kafka集群需要保证各个Broker的id在整个集群中必须唯一，需要调整这个配置项的值（如果在单机上，可以通过建立多个Broker进程来模拟分布式的Kafka集群，也需要Broker的id唯一，还需要修改一些配置目录的信息）。
在集群中的h1、h2、h3这三个节点上分别启动Kafka，分别执行如下命令：
|
1
|
bin/kafka-server-start.sh /usr/local/kafka/config/server.properties & |
可以通过查看日志，或者检查进程状态，保证Kafka集群启动成功。
我们创建一个名称为my-replicated-topic5的Topic，5个分区，并且复制因子为3，执行如下命令：
|
1
|
bin/kafka-topics.sh --create --zookeeper h1:2181,h2:2181,h3:2181/kafka --replication-factor 3 --partitions 5 --topic my-replicated-topic5 |
查看创建的Topic，执行如下命令：
|
1
|
bin/kafka-topics.sh --describe --zookeeper h1:2181,h2:2181,h3:2181/kafka --topic my-replicated-topic5 |
结果信息如下所示：
|
1
2
3
4
5
6
|
Topic:my-replicated-topic5 PartitionCount:5 ReplicationFactor:3 Configs: Topic: my-replicated-topic5 Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1 Topic: my-replicated-topic5 Partition: 1 Leader: 0 Replicas: 1,0,2 Isr: 0,2,1 Topic: my-replicated-topic5 Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,0,1 Topic: my-replicated-topic5 Partition: 3 Leader: 0 Replicas: 0,1,2 Isr: 0,2,1 Topic: my-replicated-topic5 Partition: 4 Leader: 2 Replicas: 1,2,0 Isr: 2,0,1 |
上面Leader、Replicas、Isr的含义如下：
|
1
2
3
4
|
Partition： 分区Leader ： 负责读写指定分区的节点Replicas ： 复制该分区log的节点列表Isr ： "in-sync" replicas，当前活跃的副本列表（是一个子集），并且可能成为Leader |
我们可以通过Kafka自带的bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh脚本，来验证演示如果发布消息、消费消息。
在一个终端，启动Producer，并向我们上面创建的名称为my-replicated-topic5的Topic中生产消息，执行如下脚本：
|
1
|
bin/kafka-console-producer.sh --broker-list h1:9092,h2:9092,h3:9092 --topic my-replicated-topic5 |
在另一个终端，启动Consumer，并订阅我们上面创建的名称为my-replicated-topic5的Topic中生产的消息，执行如下脚本：
|
1
|
bin/kafka-console-consumer.sh --zookeeper h1:2181,h2:2181,h3:2181/kafka --from-beginning --topic my-replicated-topic5 |
可以在Producer终端上输入字符串消息行，然后回车，就可以在Consumer终端上看到消费者消费的消息内容。
也可以参考Kafka的Producer和Consumer的Java API，通过API编码的方式来实现消息生产和消费的处理逻辑。
Storm安装配置
Storm集群也依赖Zookeeper集群，要保证Zookeeper集群正常运行。Storm的安装配置比较简单，我们仍然使用下面3台机器搭建：
|
1
2
3
|
192.168.4.142 h1192.168.4.143 h2192.168.4.144 h3 |
首先，在h1节点上，执行如下命令安装：
|
1
2
3
4
5
|
cd /usr/local/wget http://mirror.bit.edu.cn/apache/incubator/storm/apache-storm-0.9.2-incubating/apache-storm-0.9.2-incubating.tar.gztar xvzf apache-storm-0.9.2-incubating.tar.gzln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/stormchown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm |
然后，修改配置文件conf/storm.yaml，内容如下所示：
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
|
storm.zookeeper.servers: - "h1" - "h2" - "h3"storm.zookeeper.port: 2181#nimbus.host: "h1"supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703storm.local.dir: "/tmp/storm" |
将配置好的安装文件，分发到其他节点上：
|
1
2
|
scp -r /usr/local/apache-storm-0.9.2-incubating/ h2:/usr/local/scp -r /usr/local/apache-storm-0.9.2-incubating/ h3:/usr/local/ |
最后，在h2、h3节点上配置，执行如下命令：
|
1
2
3
|
cd /usr/local/ln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/stormchown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm |
Storm集群的主节点为Nimbus，从节点为Supervisor，我们需要在h1上启动Nimbus服务，在从节点h2、h3上启动Supervisor服务：
|
1
2
|
bin/storm nimbus &bin/storm supervisor & |
为了方便监控，可以启动Storm UI，可以从Web页面上监控Storm Topology的运行状态，例如在h2上启动：
|
1
|
bin/storm ui & |
这样可以通过访问http://h2:8080/来查看Topology的运行状况。
整合Kafka+Storm
消息通过各种方式进入到Kafka消息中间件，比如可以通过使用Flume来收集日志数据，然后在Kafka中路由暂存，然后再由实时计算程序Storm做实时分析，这时我们就需要将在Storm的Spout中读取Kafka中的消息，然后交由具体的Spot组件去分析处理。实际上，apache-storm-0.9.2-incubating这个版本的Storm已经自带了一个集成Kafka的外部插件程序storm-kafka，可以直接使用，例如我使用的Maven依赖配置，如下所示：
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
<dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>0.9.2-incubating</version> <scope>provided</scope></dependency><dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-kafka</artifactId> <version>0.9.2-incubating</version></dependency><dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.9.2</artifactId> <version>0.8.1.1</version> <exclusions> <exclusion> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> </exclusion> <exclusion> <groupId>log4j</groupId> <artifactId>log4j</artifactId> </exclusion> </exclusions></dependency> |
下面，我们开发了一个简单WordCount示例程序，从Kafka读取订阅的消息行，通过空格拆分出单个单词，然后再做词频统计计算，实现的Topology的代码，如下所示：
|
001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
package org.shirdrn.storm.examples;import java.util.Arrays;import java.util.HashMap;import java.util.Iterator;import java.util.Map;import java.util.Map.Entry;import java.util.concurrent.atomic.AtomicInteger;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import storm.kafka.BrokerHosts;import storm.kafka.KafkaSpout;import storm.kafka.SpoutConfig;import storm.kafka.StringScheme;import storm.kafka.ZkHosts;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.StormSubmitter;import backtype.storm.generated.AlreadyAliveException;import backtype.storm.generated.InvalidTopologyException;import backtype.storm.spout.SchemeAsMultiScheme;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.TopologyBuilder;import backtype.storm.topology.base.BaseRichBolt;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;public class MyKafkaTopology { public static class KafkaWordSplitter extends BaseRichBolt { private static final Log LOG = LogFactory.getLog(KafkaWordSplitter.class); private static final long serialVersionUID = 886149197481637894L; private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String line = input.getString(0); LOG.info("RECV[kafka -> splitter] " + line); String[] words = line.split("\\s+"); for(String word : words) { LOG.info("EMIT[splitter -> counter] " + word); collector.emit(input, new Values(word, 1)); } collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); } } public static class WordCounter extends BaseRichBolt { private static final Log LOG = LogFactory.getLog(WordCounter.class); private static final long serialVersionUID = 886149197481637894L; private OutputCollector collector; private Map<String, AtomicInteger> counterMap; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; this.counterMap = new HashMap<String, AtomicInteger>(); } @Override public void execute(Tuple input) { String word = input.getString(0); int count = input.getInteger(1); LOG.info("RECV[splitter -> counter] " + word + " : " + count); AtomicInteger ai = this.counterMap.get(word); if(ai == null) { ai = new AtomicInteger(); this.counterMap.put(word, ai); } ai.addAndGet(count); collector.ack(input); LOG.info("CHECK statistics map: " + this.counterMap); } @Override public void cleanup() { LOG.info("The final result:"); Iterator<Entry<String, AtomicInteger>> iter = this.counterMap.entrySet().iterator(); while(iter.hasNext()) { Entry<String, AtomicInteger> entry = iter.next(); LOG.info(entry.getKey() + "\t:\t" + entry.getValue().get()); } } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word", "count")); } } public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException { String zks = "h1:2181,h2:2181,h3:2181"; String topic = "my-replicated-topic5"; String zkRoot = "/storm"; // default zookeeper root configuration for storm String id = "word"; BrokerHosts brokerHosts = new ZkHosts(zks); SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id); spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme()); spoutConf.forceFromStart = false; spoutConf.zkServers = Arrays.asList(new String[] {"h1", "h2", "h3"}); spoutConf.zkPort = 2181; TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 5); // Kafka我们创建了一个5分区的Topic，这里并行度设置为5 builder.setBolt("word-splitter", new KafkaWordSplitter(), 2).shuffleGrouping("kafka-reader"); builder.setBolt("word-counter", new WordCounter()).fieldsGrouping("word-splitter", new Fields("word")); Config conf = new Config(); String name = MyKafkaTopology.class.getSimpleName(); if (args != null && args.length > 0) { // Nimbus host name passed from command line conf.put(Config.NIMBUS_HOST, args[0]); conf.setNumWorkers(3); StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology()); } else { conf.setMaxTaskParallelism(3); LocalCluster cluster = new LocalCluster(); cluster.submitTopology(name, conf, builder.createTopology()); Thread.sleep(60000); cluster.shutdown(); } }} |
上面程序，在本地调试（使用LocalCluster）不需要输入任何参数，提交到实际集群中运行时，需要传递一个参数，该参数为Nimbus的主机名称。
通过Maven构建，生成一个包含依赖的single jar文件（不要把Storm的依赖包添加进去），例如storm-examples-0.0.1-SNAPSHOT.jar，在提交Topology程序到Storm集群之前，因为用到了Kafka，需要拷贝一下依赖jar文件到Storm集群中的lib目录下面：
|
1
2
3
4
5
6
7
8
|
cp /usr/local/kafka/libs/kafka_2.9.2-0.8.1.1.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/scala-library-2.9.2.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/metrics-core-2.2.0.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/snappy-java-1.0.5.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/zkclient-0.3.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/log4j-1.2.15.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/slf4j-api-1.7.2.jar /usr/local/storm/lib/cp /usr/local/kafka/libs/jopt-simple-3.2.jar /usr/local/storm/lib/ |
然后，就可以提交我们开发的Topology程序了：
|
1
|
bin/storm jar /home/storm/storm-examples-0.0.1-SNAPSHOT.jar org.shirdrn.storm.examples.MyKafkaTopology h1 |
可以通过查看日志文件（logs/目录下）或者Storm UI来监控Topology的运行状况。如果程序没有错误，可以使用前面我们使用的Kafka Producer来生成消息，就能看到我们开发的Storm Topology能够实时接收到并进行处理。
上面Topology实现代码中，有一个很关键的配置对象SpoutConfig，配置属性如下所示：
|
1
|
spoutConf.forceFromStart = false; |
该配置是指，如果该Topology因故障停止处理，下次正常运行时是否从Spout对应数据源Kafka中的该订阅Topic的起始位置开始读取，如果forceFromStart=true，则之前处理过的Tuple还要重新处理一遍，否则会从上次处理的位置继续处理，保证Kafka中的Topic数据不被重复处理，是在数据源的位置进行状态记录。
整合Storm+HDFS
Storm实时计算集群从Kafka消息中间件中消费消息，有实时处理需求的可以走实时处理程序，还有需要进行离线分析的需求，如写入到HDFS进行分析。下面实现了一个Topology，代码如下所示：
|
001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
package org.shirdrn.storm.examples;import java.text.DateFormat;import java.text.SimpleDateFormat;import java.util.Date;import java.util.Map;import java.util.Random;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.storm.hdfs.bolt.HdfsBolt;import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;import org.apache.storm.hdfs.bolt.format.FileNameFormat;import org.apache.storm.hdfs.bolt.format.RecordFormat;import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy.TimeUnit;import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;import org.apache.storm.hdfs.bolt.sync.SyncPolicy;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.StormSubmitter;import backtype.storm.generated.AlreadyAliveException;import backtype.storm.generated.InvalidTopologyException;import backtype.storm.spout.SpoutOutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.TopologyBuilder;import backtype.storm.topology.base.BaseRichSpout;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values;import backtype.storm.utils.Utils;public class StormToHDFSTopology { public static class EventSpout extends BaseRichSpout { private static final Log LOG = LogFactory.getLog(EventSpout.class); private static final long serialVersionUID = 886149197481637894L; private SpoutOutputCollector collector; private Random rand; private String[] records; @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { this.collector = collector; rand = new Random(); records = new String[] { "10001 ef2da82d4c8b49c44199655dc14f39f6 4.2.1 HUAWEI G610-U00 HUAWEI 2 70:72:3c:73:8b:22 2014-10-13 12:36:35", "10001 ffb52739a29348a67952e47c12da54ef 4.3 GT-I9300 samsung 2 50:CC:F8:E4:22:E2 2014-10-13 12:36:02", "10001 ef2da82d4c8b49c44199655dc14f39f6 4.2.1 HUAWEI G610-U00 HUAWEI 2 70:72:3c:73:8b:22 2014-10-13 12:36:35" }; } @Override public void nextTuple() { Utils.sleep(1000); DateFormat df = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss"); Date d = new Date(System.currentTimeMillis()); String minute = df.format(d); String record = records[rand.nextInt(records.length)]; LOG.info("EMIT[spout -> hdfs] " + minute + " : " + record); collector.emit(new Values(minute, record)); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("minute", "record")); } } public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException { // use "|" instead of "," for field delimiter RecordFormat format = new DelimitedRecordFormat() .withFieldDelimiter(" : "); // sync the filesystem after every 1k tuples SyncPolicy syncPolicy = new CountSyncPolicy(1000); // rotate files FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimeUnit.MINUTES); FileNameFormat fileNameFormat = new DefaultFileNameFormat() .withPath("/storm/").withPrefix("app_").withExtension(".log"); HdfsBolt hdfsBolt = new HdfsBolt() .withFileNameFormat(fileNameFormat) .withRecordFormat(format) .withRotationPolicy(rotationPolicy) .withSyncPolicy(syncPolicy); TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("event-spout", new EventSpout(), 3); builder.setBolt("hdfs-bolt", hdfsBolt, 2).fieldsGrouping("event-spout", new Fields("minute")); Config conf = new Config(); String name = StormToHDFSTopology.class.getSimpleName(); if (args != null && args.length > 0) { conf.put(Config.NIMBUS_HOST, args[0]); conf.setNumWorkers(3); StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology()); } else { conf.setMaxTaskParallelism(3); LocalCluster cluster = new LocalCluster(); cluster.submitTopology(name, conf, builder.createTopology()); Thread.sleep(60000); cluster.shutdown(); } }} |
上面的处理逻辑，可以对HdfsBolt进行更加详细的配置，如FileNameFormat、SyncPolicy、FileRotationPolicy（可以设置在满足什么条件下，切出一个新的日志，如可以指定多长时间切出一个新的日志文件，可以指定一个日志文件大小达到设置值后，再写一个新日志文件），更多设置可以参考storm-hdfs，。
上面代码在打包的时候，需要注意，使用storm-starter自带的Maven打包配置，可能在将Topology部署运行的时候，会报错，可以使用maven-shade-plugin这个插件，如下配置所示：
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>1.4</version> <configuration> <createDependencyReducedPom>true</createDependencyReducedPom> </configuration> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass></mainClass> </transformer> </transformers> </configuration> </execution> </executions></plugin> |
整合Kafka+Storm+HDFS
上面分别对整合Kafka+Storm和Storm+HDFS做了实践，可以将后者的Spout改成前者的Spout，从Kafka中消费消息，在Storm中可以做简单处理，然后将数据写入HDFS，最后可以在Hadoop平台上对数据进行离线分析处理。下面，写了一个简单的例子，从Kafka消费消息，然后经由Storm处理，写入到HDFS存储，代码如下所示：
|
001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
package org.shirdrn.storm.examples;import java.util.Arrays;import java.util.Map;import org.apache.commons.logging.Log;import org.apache.commons.logging.LogFactory;import org.apache.storm.hdfs.bolt.HdfsBolt;import org.apache.storm.hdfs.bolt.format.DefaultFileNameFormat;import org.apache.storm.hdfs.bolt.format.DelimitedRecordFormat;import org.apache.storm.hdfs.bolt.format.FileNameFormat;import org.apache.storm.hdfs.bolt.format.RecordFormat;import org.apache.storm.hdfs.bolt.rotation.FileRotationPolicy;import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy;import org.apache.storm.hdfs.bolt.rotation.TimedRotationPolicy.TimeUnit;import org.apache.storm.hdfs.bolt.sync.CountSyncPolicy;import org.apache.storm.hdfs.bolt.sync.SyncPolicy;import storm.kafka.BrokerHosts;import storm.kafka.KafkaSpout;import storm.kafka.SpoutConfig;import storm.kafka.StringScheme;import storm.kafka.ZkHosts;import backtype.storm.Config;import backtype.storm.LocalCluster;import backtype.storm.StormSubmitter;import backtype.storm.generated.AlreadyAliveException;import backtype.storm.generated.InvalidTopologyException;import backtype.storm.spout.SchemeAsMultiScheme;import backtype.storm.task.OutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.TopologyBuilder;import backtype.storm.topology.base.BaseRichBolt;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;public class DistributeWordTopology { public static class KafkaWordToUpperCase extends BaseRichBolt { private static final Log LOG = LogFactory.getLog(KafkaWordToUpperCase.class); private static final long serialVersionUID = -5207232012035109026L; private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String line = input.getString(0).trim(); LOG.info("RECV[kafka -> splitter] " + line); if(!line.isEmpty()) { String upperLine = line.toUpperCase(); LOG.info("EMIT[splitter -> counter] " + upperLine); collector.emit(input, new Values(upperLine, upperLine.length())); } collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("line", "len")); } } public static class RealtimeBolt extends BaseRichBolt { private static final Log LOG = LogFactory.getLog(KafkaWordToUpperCase.class); private static final long serialVersionUID = -4115132557403913367L; private OutputCollector collector; @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; } @Override public void execute(Tuple input) { String line = input.getString(0).trim(); LOG.info("REALTIME: " + line); collector.ack(input); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { } } public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException { // Configure Kafka String zks = "h1:2181,h2:2181,h3:2181"; String topic = "my-replicated-topic5"; String zkRoot = "/storm"; // default zookeeper root configuration for storm String id = "word"; BrokerHosts brokerHosts = new ZkHosts(zks); SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, zkRoot, id); spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme()); spoutConf.forceFromStart = false; spoutConf.zkServers = Arrays.asList(new String[] {"h1", "h2", "h3"}); spoutConf.zkPort = 2181; // Configure HDFS bolt RecordFormat format = new DelimitedRecordFormat() .withFieldDelimiter("\t"); // use "\t" instead of "," for field delimiter SyncPolicy syncPolicy = new CountSyncPolicy(1000); // sync the filesystem after every 1k tuples FileRotationPolicy rotationPolicy = new TimedRotationPolicy(1.0f, TimeUnit.MINUTES); // rotate files FileNameFormat fileNameFormat = new DefaultFileNameFormat() .withPath("/storm/").withPrefix("app_").withExtension(".log"); // set file name format HdfsBolt hdfsBolt = new HdfsBolt() .withFileNameFormat(fileNameFormat) .withRecordFormat(format) .withRotationPolicy(rotationPolicy) .withSyncPolicy(syncPolicy); // configure & build topology TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 5); builder.setBolt("to-upper", new KafkaWordToUpperCase(), 3).shuffleGrouping("kafka-reader"); builder.setBolt("hdfs-bolt", hdfsBolt, 2).shuffleGrouping("to-upper"); builder.setBolt("realtime", new RealtimeBolt(), 2).shuffleGrouping("to-upper"); // submit topology Config conf = new Config(); String name = DistributeWordTopology.class.getSimpleName(); if (args != null && args.length > 0) { String nimbus = args[0]; conf.put(Config.NIMBUS_HOST, nimbus); conf.setNumWorkers(3); StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology()); } else { conf.setMaxTaskParallelism(3); LocalCluster cluster = new LocalCluster(); cluster.submitTopology(name, conf, builder.createTopology()); Thread.sleep(60000); cluster.shutdown(); } }} |
上面代码中，名称为to-upper的Bolt将接收到的字符串行转换成大写以后，会将处理过的数据向后面的hdfs-bolt、realtime这两个Bolt各发一份拷贝，然后由这两个Bolt分别根据实际需要（实时/离线）单独处理。
打包后，在Storm集群上部署并运行这个Topology：
|
1
|
bin/storm jar ~/storm-examples-0.0.1-SNAPSHOT.jar org.shirdrn.storm.examples.DistributeWordTopology h1 |
可以通过Storm UI查看Topology运行情况，可以查看HDFS上生成的数据。
参考链接
- http://kafka.apache.org/
- http://kafka.apache.org/documentation.html
- https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
- http://storm.apache.org/
- http://storm.apache.org/documentation/Tutorial.html
- http://storm.apache.org/documentation/FAQ.html
- https://github.com/ptgoetz/storm-hdfs
转载请注明：SuperIT » Kafka+Storm+HDFS整合实践
