Storm和Kafka的整合安裝和測試
1、先保證Storm叢集已經安裝
Storm叢集的安裝可以參考文章：Storm叢集的安裝，Kafka的安裝和測試也可以參考apache kafka官網上的Quick Start
2、安裝Kafka叢集：
2.1 下載Kafka包，我們這裡選擇kafka_2.9.2-0.8.1.tgz版本進行安裝，解壓到安裝目錄：tar -xzf kafka_2.9.2-0.8.1.tgz
2.2 建立一個軟連結kafka：ln -s kafka_2.9.2-0.8.1 kafka，方便以後升級
2.3 修改~/.profile，匯出KAFKA_HOME和PATH
export KAFKA_HOME=/home/storm/software/kafka
export PATH=.:$KAFKA_HOME/bin:$PATH
2.4 執行：source ~/.profile，使得修改對於當前會話生效
2.5 修改kafka配置檔案kafka/config/server.properties：
log.dirs=/home/storm/software/kafka/kafka-logs
zookeeper.connect=linux-21:2181,linux-7:2181
2.6 啟動kafka：nohup kafka-server-start.sh ./config/server.properties &
檢視日誌cat nohup.out，有可能會報如下錯誤：
Unrecognized VM option '+UseCompressedOops'
Could not create the Java virtual machine.
這是由於機器上安裝的JDK版本不支援VM選項導致的，Kafka0.8.1版本需要的JDK是JDK 1.7 u51。可以通過如下方法暫時規避，但是是否還有其他相容性問題，還不確定：修改kafka/bin目錄下的KAFKA_JVM_PERFORMANCE_OPTS取值，把-XX:+UseCompressedOops刪除
2.7 建立topic：
kafka-topics.sh --create --zookeeper linux-7:2181 --partitions 2 --replication-factor 1 --topic test
其中--zookeeper指定一個zookeeper的位置，
--partition指定partition的個數，建立完後，我們能在log.dirs指定的目錄下看到兩個以主題名稱建立的資料夾：test-0,test-1，裡面有一個索引檔案.index，一個數據檔案.log
--relication-factor指定副本的個數(這個值不能大於broker的個數)
--topic指定主題的名字
2.8 檢視topic：
kafka-topics.sh --list --zookeeper linux-21:2181
2.9 produce訊息：
kafka-console-producer.sh --broker-list localhost:9092 --topic test
--broker-list指定broker的列表
2.10 consume訊息：
kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
--zookeeper指定zookeeper的位置
我們從produce命令和consume命令可以看出，produce是直接和broker通訊的；而consume不需要和broker直接通訊，它是從zookeeper通訊而獲得broker的資訊的。
2.11 將整個kafka目錄拷貝到其他kafka叢集機器上，注意~/.profile的修改和生效，以及修改server.properties檔案中的broker.id屬性，使得所有叢集的機器取值不同。
2,12 將其他機器上的kafka啟動，並測試producer和consumer指令。
3、編寫kafka Producer JAVA程式
3.1 先構造一個ProducerConfig物件，用來配置kafka broker的資訊
3.2 構造Producer，呼叫send命令傳送KeyedMessage訊息。
3.3 這裡要注意兩個問題：
- 程式碼中的API已經過時，已經有新的API替換
- 如果你寫的Producer程式是在本機(非kafka伺服器) 上執行，那麼一定要配置好server.properties配置檔案的advertised.host.name引數，配置檔案中關於這個引數有如下說明：
# Hostname the broker will advertise to producers and consumers. If not set, it uses the
# value for "host.name" if configured. Otherwise, it will use the value returned from
# java.net.InetAddress.getCanonicalHostName().意思就是說廣播給producer和consumer用的。如果沒有設定，就去讀host.name中的設定，如果也沒設定，就通過getCanonicalHostName()來取。所以如果你的本機沒有配置hosts檔案的話，就無法解析發過來的broker主機名。如果不想修改本機的hosts檔案，直接把這個引數配置成broker的IP地址就可以了。
程式碼如下：
package com.mykafka.producer; import java.io.BufferedReader; import java.io.FileReader; import java.io.IOException; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; public class MyLogProducer { public static void main(String[] args) throws IOException { Properties props = new Properties(); props.put("metadata.broker.list", "10.118.15.7:9092"); props.put("serializer.class", "kafka.serializer.StringEncoder"); //props.put("partitioner.class", "example.producer.SimplePartitioner"); props.put("request.required.acks", "1"); props.put("producer.type", "async"); props.put("compression.codec", "1"); //props.put("zookeeper.connect", "linux-7:2181,linux-21:2181"); ProducerConfig config = new ProducerConfig(props); Producer<String, String> producer = new Producer<String, String>(config); BufferedReader reader = new BufferedReader(new FileReader("c:/access.log"), 2048); String aline; while((aline = reader.readLine()) != null){ System.out.println(aline); KeyedMessage<String, String> data = new KeyedMessage<String, String>("logs", aline); producer.send(data); } producer.close(); reader.close(); System.out.println("send over...."); } }
可以在broker端開一個consumer來檢查是否已經收到這段程式發過來的資料。
4、Storm consume Kafka
Storm從Kafka讀取資料，我們可以利用一個開源的jar包來實現。
主要分兩步來使用：
第一步，構造一個SpoutConfig物件，它需要zookeeper的資訊，以及topic資訊等。
第二步，用SpoutConfig來構造KafkaSpout物件
import com.storm.stormkafka08.bolt.LogParserBolt; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.spout.SchemeAsMultiScheme; import backtype.storm.topology.TopologyBuilder; import storm.kafka.KafkaSpout; import storm.kafka.SpoutConfig; import storm.kafka.StringScheme; import storm.kafka.ZkHosts; public class LogTopology { public static void main(String[] args) { // zookeeper hosts for the Kafka cluster ZkHosts zkHosts = new ZkHosts("10.118.15.21:2181"); // Create the KafkaSpout configuration // Second argument is the topic name // Third argument is the ZooKeeper root for Kafka // Fourth argument is consumer group id SpoutConfig kafkaConfig = new SpoutConfig(zkHosts, "logs", "", "id7"); // Specify that the kafka messages are String kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); // We want to consume all the first messages in // the topic every time we run the topology to // help in debugging. In production, this // property should be false kafkaConfig.forceFromStart = true; // Now we create the topology TopologyBuilder builder = new TopologyBuilder(); // set the kafka spout class builder.setSpout("KafkaSpout", new KafkaSpout(kafkaConfig), 1); // configure the bolts builder.setBolt("ParseLog", new LogParserBolt(), 3).shuffleGrouping("KafkaSpout"); // create an instance of LocalCluster class // for executing topology in local mode. LocalCluster cluster = new LocalCluster(); Config conf = new Config(); // Submit topology for execution cluster.submitTopology("KafkaToplogy", conf, builder.createTopology()); try { // Wait for some time before exiting System.out.println("Waiting to consume from kafka"); Thread.sleep(10000); } catch (Exception exception) { System.out.println("Thread interrupted exception : " + exception); } // kill the KafkaTopology cluster.killTopology("KafkaToplogy"); // shut down the storm test cluster cluster.shutdown(); } }