Kafka+Storm+HDFS整合实践

Posted 2020-07-28 专注it

在基于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	192.168.4.142   h1

2	192.168.4.143   h2

3	192.168.4.144   h3

在安装Kafka集群之前，这里没有使用Kafka自带的Zookeeper，而是独立安装了一个Zookeeper集群，也是使用这3台机器，保证Zookeeper集群正常运行。
首先，在h1上准备Kafka安装文件，执行如下命令：

1	cd /usr/local/

2	wget http://mirror.bit.edu.cn/apache/kafka/0.8.1.1/kafka_2.9.2-0.8.1.1.tgz

3	tar xvzf kafka_2.9.2-0.8.1.1.tgz

4	ln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

5	chown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

修改配置文件/usr/local/kafka/config/server.properties，修改如下内容：

1	broker.id=0

2	zookeeper.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	cd /usr/local/zookeeper

2	bin/zkCli.sh

在ZooKeeper执行如下命令创建chroot路径：

1	create /kafka ‘‘

这样，每次连接Kafka集群的时候（使用--zookeeper选项），也必须使用带chroot路径的连接字符串，后面会看到。
然后，将配置好的安装文件同步到其他的h2、h3节点上：

1	scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h2:/usr/local/

2	scp -r /usr/local/kafka_2.9.2-0.8.1.1/ h3:/usr/local/

最后，在h2、h3节点上配置，执行如下命令：

1	cd /usr/local/

2	ln -s /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

3	chown -R kafka:kafka /usr/local/kafka_2.9.2-0.8.1.1 /usr/local/kafka

并修改配置文件/usr/local/kafka/config/server.properties内容如下所示：

1	broker.id=1  # 在h1修改

2

3	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	Topic:my-replicated-topic5     PartitionCount:5     ReplicationFactor:3     Configs:

2	Topic: my-replicated-topic5     Partition: 0     Leader: 0     Replicas: 0,2,1     Isr: 0,2,1

3	Topic: my-replicated-topic5     Partition: 1     Leader: 0     Replicas: 1,0,2     Isr: 0,2,1

4	Topic: my-replicated-topic5     Partition: 2     Leader: 2     Replicas: 2,1,0     Isr: 2,0,1

5	Topic: my-replicated-topic5     Partition: 3     Leader: 0     Replicas: 0,1,2     Isr: 0,2,1

6	Topic: my-replicated-topic5     Partition: 4     Leader: 2     Replicas: 1,2,0     Isr: 2,0,1

上面Leader、Replicas、Isr的含义如下：

1	Partition： 分区

2	Leader   ： 负责读写指定分区的节点

3	Replicas ： 复制该分区log的节点列表

4	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	192.168.4.142   h1

2	192.168.4.143   h2

3	192.168.4.144   h3

首先，在h1节点上，执行如下命令安装：

1	cd /usr/local/

2	wget http://mirror.bit.edu.cn/apache/incubator/storm/apache-storm-0.9.2-incubating/apache-storm-0.9.2-incubating.tar.gz

3	tar xvzf apache-storm-0.9.2-incubating.tar.gz

4	ln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

5	chown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

然后，修改配置文件conf/storm.yaml，内容如下所示：

01	storm.zookeeper.servers:

02

- "h1"

03

- "h2"

04

- "h3"

05	storm.zookeeper.port: 2181

06

#

07	nimbus.host: "h1"

08

09	supervisor.slots.ports:

10

- 6700

11

- 6701

12

- 6702

13

- 6703

14

15	storm.local.dir: "/tmp/storm"

将配置好的安装文件，分发到其他节点上：

1	scp -r /usr/local/apache-storm-0.9.2-incubating/ h2:/usr/local/

2	scp -r /usr/local/apache-storm-0.9.2-incubating/ h3:/usr/local/

最后，在h2、h3节点上配置，执行如下命令：

1	cd /usr/local/

2	ln -s /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

3	chown -R storm:storm /usr/local/apache-storm-0.9.2-incubating /usr/local/storm

Storm集群的主节点为Nimbus，从节点为Supervisor，我们需要在h1上启动Nimbus服务，在从节点h2、h3上启动Supervisor服务：

1	bin/storm nimbus &

2	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	<dependency>

02	<groupId>org.apache.storm</groupId>

03	<artifactId>storm-core</artifactId>

04	<version>0.9.2-incubating</version>

05	<scope>provided</scope>

06	</dependency>

07	<dependency>

08	<groupId>org.apache.storm</groupId>

09	<artifactId>storm-kafka</artifactId>

10	<version>0.9.2-incubating</version>

11	</dependency>

12	<dependency>

13	<groupId>org.apache.kafka</groupId>

14	<artifactId>kafka_2.9.2</artifactId>

15	<version>0.8.1.1</version>

16	<exclusions>

17	<exclusion>

18	<groupId>org.apache.zookeeper</groupId>

19	<artifactId>zookeeper</artifactId>

20	</exclusion>

21	<exclusion>

22	<groupId>log4j</groupId>

23	<artifactId>log4j</artifactId>

24	</exclusion>

25	</exclusions>

26	</dependency>

下面，我们开发了一个简单WordCount示例程序，从Kafka读取订阅的消息行，通过空格拆分出单个单词，然后再做词频统计计算，实现的Topology的代码，如下所示：

001	package org.shirdrn.storm.examples;

002

003	import java.util.Arrays;

004	import java.util.HashMap;

005	import java.util.Iterator;

006	import java.util.Map;

007	import java.util.Map.Entry;

008	import java.util.concurrent.atomic.AtomicInteger;

009

010	import org.apache.commons.logging.Log;

011	import org.apache.commons.logging.LogFactory;

012

013	import storm.kafka.BrokerHosts;

014	import storm.kafka.KafkaSpout;

015	import storm.kafka.SpoutConfig;

016	import storm.kafka.StringScheme;

017	import storm.kafka.ZkHosts;

018	import backtype.storm.Config;

019	import backtype.storm.LocalCluster;

020	import backtype.storm.StormSubmitter;

021	import backtype.storm.generated.AlreadyAliveException;

022	import backtype.storm.generated.InvalidTopologyException;

023	import backtype.storm.spout.SchemeAsMultiScheme;

024	import backtype.storm.task.OutputCollector;

025	import backtype.storm.task.TopologyContext;

026	import backtype.storm.topology.OutputFieldsDeclarer;

027	import backtype.storm.topology.TopologyBuilder;

028	import backtype.storm.topology.base.BaseRichBolt;

029	import backtype.storm.tuple.Fields;

030	import backtype.storm.tuple.Tuple;

031	import backtype.storm.tuple.Values;

032

033	public class MyKafkaTopology {

034

035	public static class KafkaWordSplitter extends BaseRichBolt {

036

037	private static final Log LOG = LogFactory.getLog(KafkaWordSplitter.class);

038	private static final long serialVersionUID = 886149197481637894L;

039	private OutputCollector collector;

040

041

@Override

042	public void prepare(Map stormConf, TopologyContext context,

043	OutputCollector collector) {

044	this.collector = collector;

045

}

046

047

@Override

048	public void execute(Tuple input) {

049	String line = input.getString(0);

050	LOG.info("RECV[kafka -> splitter] " + line);

051	String[] words = line.split("\\s+");

052	for(String word : words) {

053	LOG.info("EMIT[splitter -> counter] " + word);

054	collector.emit(input, new Values(word, 1));

055

}

056	collector.ack(input);

057

}

058

059

@Override

060	public void declareOutputFields(OutputFieldsDeclarer declarer) {

061	declarer.declare(new Fields("word", "count"));

062

}

063

064

}

065

066	public static class WordCounter extends BaseRichBolt {

067

068	private static final Log LOG = LogFactory.getLog(WordCounter.class);

069	private static final long serialVersionUID = 886149197481637894L;

070	private OutputCollector collector;

071	private Map<String, AtomicInteger> counterMap;

072

073

@Override

074	public void prepare(Map stormConf, TopologyContext context,

075	OutputCollector collector) {

076	this.collector = collector;

077	this.counterMap = new HashMap<String, AtomicInteger>();

078

}

079

080

@Override

081	public void execute(Tuple input) {

082	String word = input.getString(0);

083	int count = input.getInteger(1);

084	LOG.info("RECV[splitter -> counter] " + word + " : " + count);

085	AtomicInteger ai = this.counterMap.get(word);

086	if(ai == null) {

087	ai = new AtomicInteger();

088	this.counterMap.put(word, ai);

089

}

090	ai.addAndGet(count);

091	collector.ack(input);

092	LOG.info("CHECK statistics map: " + this.counterMap);

093

}

094

095

@Override

096	public void cleanup() {

097	LOG.info("The final result:");

098	& 以上是关于Kafka+Storm+HDFS整合实践的主要内容，如果未能解决你的问题，请参考以下文章 Kafka+Storm+HDFS 整合示例 PB级大数据平台Kafka + HDFS整合与资源评估实践 SpringBoot整合Kafka和Storm SpringBoot整合Kafka和Storm storm 整合 kafka之保存MySQL数据库 storm和kafka整合