【JAVA版】Storm程序整合Kafka、Mongodb示例及部署

一、環境

• 一台Centos6.5主機
• Mongo 3.0
• Kafka_2.11-0.8.2.1
• Storm-0.9.5
• Zookeeper-3.4.6
• java 1.7 （后因在mac上打包的jar由1.8編譯沒法兒運行，改為java 1.8）
• 其余環境暫略

二、運行啟動

1. 啟動zookeeper

   確認配置正確，配置相關可自行搜索。

   [root@localhost zookeeper-3.4.6]#pwd
   /data0/xxx/zookeeper-3.4.6
   [root@localhost zookeeper-3.4.6]#bin/zkServer.sh start

   這里路徑在zookeeper根目錄啟動，主要是這樣運行會自動在當前目錄生成一個日志文件。也就是說在哪個目錄啟動，日志文件就在哪兒。這是為了將日志文件放在該目錄的一種做法。

2. 啟動kafka

   /data0/xxx/kafka_2.11-0.8.2.1/bin/kafka-server-start.sh /data0/xxx/kafka_2.11-0.8.2.1/config/server-0.properties > /data0/xxx/kafka_2.11-0.8.2.1/logs/server-0.log 2>&1 &
   /data0/xxx/kafka_2.11-0.8.2.1/bin/kafka-server-start.sh /data0/xxx/kafka_2.11-0.8.2.1/config/server-1.properties > /data0/xxx/kafka_2.11-0.8.2.1/logs/server-1.log 2>&1 &
   /data0/xxx/kafka_2.11-0.8.2.1/bin/kafka-server-start.sh /data0/xxx/kafka_2.11-0.8.2.1/config/server-2.properties > /data0/xxx/kafka_2.11-0.8.2.1/logs/server-2.log 2>&1 &

   這里我在單機上起了3個kafka進程，作為單機集群。
   配置0示例：

   broker.id=0
   port=9092
   host.name=172.16.0.100
   num.network.threads=3
   num.io.threads=8
   socket.send.buffer.bytes=102400
   socket.receive.buffer.bytes=102400
   socket.request.max.bytes=104857600
   log.dirs=/data0/xxx/kafka_2.11-0.8.2.1/log_0
   num.partitions=1
   num.recovery.threads.per.data.dir=1
   log.retention.hours=168
   log.segment.bytes=1073741824
   log.retention.check.interval.ms=300000
   log.cleaner.enable=false
   zookeeper.connect=localhost:2181
   zookeeper.connection.timeout.ms=6000

3. 啟動storm

   /data0/xxx/storm-0.9.5/bin/storm nimbus > /data0/xxx/storm-0.9.5/log/nimbus.log 2>&1 &
   /data0/xxx/storm-0.9.5/bin/storm supervisor > /data0/xxx/storm-0.9.5/log/supervisor.log 2>&1 &
   /data0/xxx/storm-0.9.5/bin/storm ui > /data0/xxx/storm-0.9.5/log/ui.log 2>&1 &
   /data0/xxx/storm-0.9.5/bin/storm logviewer > /data0/xxx/storm-0.9.5/log/logviewer.log 2>&1 &

   依次啟動nimbus、supervisor、ui、logviewer
   storm配置文件示例：

   storm.zookeeper.servers:
       - "172.16.0.100"
   nimbus.host: "172.16.0.100"
   storm.local.dir: "/data0/xxx/storm-0.9.5/workdir"
   storm.messaging.netty.max_retries: 30
   storm.messaging.netty.min_wait_ms: 100
   storm.messaging.netty.max_wait_ms: 1500
   supervisor.slots.ports:
     - 6700
     - 6701
     - 6702
     - 6703
   supervisor.worker.start.timeout.secs: 60

三、初始化設置

    在kafka中創建topic，配置數據源導入kafka等等略。另外還有mongodb的初始化設置，建索引等。

四、storm程序

• 環境：mac air、Eclipse IDE for Java Developers（這是我用的版本，需要maven）
• 代碼示例
• 提交運行

  /path/bin/storm jar ds.nginx-0.0.1-SNAPSHOT-jar-with-dependencies.jar cms.ds.nginx.LogTopology ds_nginx_topology remote