【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

   确认配置正确，配置相关可自行搜索。

   [[email protected] zookeeper-3.4.6]#pwd
   /data0/xxx/zookeeper-3.4.6
   [[email protected] 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

   server-0.properties

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

   storm.yaml

三、初始化设置

在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