Storm读取Kafka数据是如何实现的

问题导读

1.本文基于什么版本？
2.Storm读取Kafka数据是如何实现的？
3.实现一个Kafka Spout有哪两种方式？

Strom整合Kafka版本信息

Storm与Kafka的版本信息：

Storm：apache-storm-0.9.2-incubating
Kafka：kafka_2.9.2-0.8.1.1.tgz

Strom从Kafka中读取数据本质

实现Storm读取Kafka中的数据，参考官网介绍，本部分主要参考自storm-kafka的README。

Strom从Kafka中读取数据，本质：实现一个Storm中的Spout，来读取Kafka中的数据；这个Spout，可以称为Kafka Spout。实现一个Kafka Spout有两条路：

core storm spout；
Trident spout；

无论用哪种方式实现Kafka Spout，都分为两步走：

实现BrokerHost接口：用于记录Kafka broker host与partition之间的映射关系；具体两种实现方式：
- ZkHosts类：从zookeeper中动态的获取kafka broker与partition之间的映射关系；初始化时，需要配置zookeeper的ip:port；默认，每60s从zookeeper中请求一次映射关系；
- StaticHosts类：当broker–partition之间的映射关系是静态时，常使用此方法；
继承KafkaConfig类：用于存储Kafka相关的参数；将上面实例的BrokerHost对象，作为参数传入KafkaConfig，例，Kafka的一个构造方法为KafkaConfig(BrokerHosts hosts, String topic)；当前其实现方式有两个：
- SpoutConfig：Core KafkaSpout只接受此配置方式；
- TridentKafkaConfig：TridentKafkaEmitter只接受此配置方式；

KafkaConfig类中涉及到的配置参数默认值如下：

[mw_shl_code=bash,true]public int fetchSizeBytes = 1024 * 1024;
public int socketTimeoutMs = 10000;
public int fetchMaxWait = 10000;
public int bufferSizeBytes = 1024 * 1024;
public MultiScheme scheme = new RawMultiScheme();
public boolean forceFromStart = false;
public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
public long maxOffsetBehind = Long.MAX_VALUE;
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
public int metricsTimeBucketSizeInSecs = 60;[/mw_shl_code]

上面的MultiScheme类型的参数shceme，其负责：将Kafka中取出的byte[]转换为storm所需的tuple，这是一个扩展点，默认是原文输出。两种实现：SchemeAsMultiScheme和KeyValueSchemeAsMultiScheme可将读取的byte[]转换为String。

notes(ningg)：几个疑问，列在下面了

ZkHosts类的一个构造方法ZkHosts(String brokerZkStr, String brokerZkPath)，其中brokerZkPath的含义，原始给出的说法是：”rokerZkPath is the root directory under which all the topics and partition information is stored. by Default this is /brokers which is what default kafka implementation uses.”
SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id)，其中，zkRoot是一个root目录，用于存储consumer的offset；那这个zkRoot对应的目录物理上在哪台机器？

配置实例Core Kafka Spout

本质是设置一个读取Kafka中数据的Kafka Spout，然后，将从替换原始local mode下，topology中的Spout即可。下面是一个已经验证过的实例

[mw_shl_code=bash,true]TopologyBuilder builder = new TopologyBuilder();
BrokerHosts hosts = new ZkHosts("121.7.2.12:2181");
SpoutConfig spoutConfig = new SpoutConfig(hosts, "ningg", "/" + "ningg", UUID.randomUUID().toString());
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
// set Spout.
builder.setSpout("word", kafkaSpout, 3);
builder.setBolt("result", new ExclamationBolt(), 3).shuffleGrouping("word");
Config conf = new Config();
conf.setDebug(true);
// submit topology in local mode
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());[/mw_shl_code]

Trident Kafka Spout（todo）

todo

下面的样例并还没验证：
[mw_shl_code=bash,true]TridentTopology topology = new TridentTopology();
BrokerHosts zk = new ZkHosts("localhost");
TridentKafkaConfig spoutConf = new TridentKafkaConfig(zk, "test-topic");
spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
OpaqueTridentKafkaSpout spout = new OpaqueTridentKafkaSpout(spoutConf);[/mw_shl_code]