storm-kafka 入門

摘要： 在資料密集型的系統架構中，一般會使用Kafka作為中間的緩衝，使用Storm做資料統計時，我們一般會從Kafka中消費訊息，然後把最終的統計結果儲存到Redis中，storm-kafka模組極大方便了我們在開發Spout時和Kafka的整合。 本文先簡單介紹下KafkaSpou...

在資料密集型的系統架構中，一般會使用Kafka作為中間的緩衝，使用Storm做資料統計時，我們一般會從Kafka中消費訊息，然後把最終的統計結果儲存到Redis中，storm-kafka模組極大方便了我們在開發Spout時和Kafka的整合。

本文先簡單介紹下KafkaSpout，然後在前面 ofollow,noindex" target="_blank">WordCountRedisTopology 的基礎上引入storm-kafka。

KafkaSpout

在構造KafkaSpout的時候需要指定一個KafkaConfig。

public class KafkaConfig implements Serializable {
private static final long serialVersionUID = 5276718734571623855L;

public final BrokerHosts hosts;
public final String topic;
public final String clientId;

public int fetchSizeBytes = 1024 * 1024;
public int socketTimeoutMs = 10000;
public int fetchMaxWait = FetchRequest.DefaultMaxWait();
public int bufferSizeBytes = 1024 * 1024;
public MultiScheme scheme = new RawMultiScheme();
public boolean ignoreZkOffsets = false;
public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
public long maxOffsetBehind = Long.MAX_VALUE;
public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
public int metricsTimeBucketSizeInSecs = 60;
public int minFetchByte = FetchRequest.DefaultMinBytes();

public KafkaConfig(BrokerHosts hosts, String topic) {
this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());
}

public KafkaConfig(BrokerHosts hosts, String topic, String clientId) {
this.hosts = hosts;
this.topic = topic;
this.clientId = clientId;
}
}

其中的BrokerHosts ，指定了Kafka Broker的地址。實現類ZkHosts表示Broker的地址從Kafka中動態獲取。

topic指定了要消費的主題，clientId用於ZK路徑中儲存KafkaSpout的偏移量。

項scheme用於配置從Kafka消費的ByteBuffer如何轉換為Storm裡面的Tuple，及控制Spout輸出欄位的命名，下面示例中會看到。

KafkaConfig子類SpoutConfig 提供了更多的控制，比如設定ZK的根路徑。

public class SpoutConfig extends KafkaConfig implements Serializable {
public List<String> zkServers = null;
public Integer zkPort = null;
public String zkRoot = null;
public String id = null;

public String outputStreamId;

// setting for how often to save the current kafka offset to ZooKeeper
public long stateUpdateIntervalMs = 2000;

// Retry strategy for failed messages
public String failedMsgRetryManagerClass = ExponentialBackoffMsgRetryManager.class.getName();

// Exponential back-off retry settings.These are used by ExponentialBackoffMsgRetryManager for retrying messages after a bolt
// calls OutputCollector.fail(). These come into effect only if ExponentialBackoffMsgRetryManager is being used.
public long retryInitialDelayMs = 0;
public double retryDelayMultiplier = 1.0;
public long retryDelayMaxMs = 60 * 1000;
public int retryLimit = -1;

/**
* Create a SpoutConfig without setting client.id, which can make the source application ambiguous when tracing Kafka calls.
*/
public SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id) {
super(hosts, topic);
this.zkRoot = zkRoot;
this.id = id;
}

/**
* Create a SpoutConfig with a client.id value.
*/
public SpoutConfig(BrokerHosts hosts, String topic, String clientId, String zkRoot, String id) {
super(hosts, topic, clientId);
this.zkRoot = zkRoot;
this.id = id;
}
}

示例

主類中設定SpoutConfig及拓撲結構。

BrokerHosts zkrHosts = new ZkHosts(PropertiesUtils.getKafkaProp("kafka.zkhosts"));
final String kafkaTopic = PropertiesUtils.getKafkaProp("kafka.topic");
final String zkRoot = "/" + kafkaTopic;
final String spoutId = kafkaTopic;

SpoutConfig kafkaConfig = new SpoutConfig(zkrHosts, kafkaTopic, zkRoot, spoutId);
// default is byte arr
kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("spout", new KafkaSpout(kafkaConfig), 5);
builder.setBolt("split", new SplitSentenceBolt2(), 8).shuffleGrouping("spout");
builder.setBolt("count", new WordCountRedisBolt(jedisPoolConfig), 12).fieldsGrouping("split", new Fields("word"));

Config conf = new Config();
conf.setDebug(true);

if (args != null && args.length > 0) {
conf.setNumWorkers(3);
StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
}

注意，這裡自定義了配置中的scheme，這樣我們的tuple會作為一個字串傳遞，並且輸出欄位名是str。

public class StringScheme implements Scheme {
private static final Charset UTF8_CHARSET = StandardCharsets.UTF_8;
public static final String STRING_SCHEME_KEY = "str";

public List<Object> deserialize(ByteBuffer bytes) {
return new Values(deserializeString(bytes));
}

public static String deserializeString(ByteBuffer string) {
if (string.hasArray()) {
int base = string.arrayOffset();
return new String(string.array(), base + string.position(), string.remaining());
} else {
return new String(Utils.toByteArray(string), UTF8_CHARSET);
}
}

public Fields getOutputFields() {
return new Fields(STRING_SCHEME_KEY);
}
}

後續的Bolt就可以接到對應的Tuple。

public class SplitSentenceBolt2 extends BaseBasicBolt {
@Override
public void execute(Tuple input, BasicOutputCollector collector) {
String sentence = input.getStringByField("str");
String[] words = sentence.split(" ");
for (String w : words) {
collector.emit(new Values(w));
}
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}

使用kafka命令列工具構造一個訊息。

➜kafka_2.11-2.0.0 ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>x
>hello
>hello
>kafka
>kafka is what
>reading is a good habit
>kafka kafka broker partition 
>reading is a good habit
>storm is coming
>let it come storm
>hello
>hello

然後可以在redis中看到統計結果。

127.0.0.1:6379> scan 0 match cnt.*
1) "5"
2)1) "cnt.x"
2) "cnt.good"
3) "cnt.it"
4) "cnt.let"
5) "cnt.habit"
6) "cnt.partition"
7) "cnt.coming"
8) "cnt.what"
9) "cnt.is"
10) "cnt.kafka"
127.0.0.1:6379> get cnt.storm
"2"
127.0.0.1:6379> get cnt.kafka
"4"
127.0.0.1:6379>

相關閱讀

http:// storm.apache.org/releas es/1.0.6/storm-kafka.html

文中程式碼

原文地址