spark读取 kafka nginx网站日志消息 并写入HDFS中

spark 版本为1.0

kafka 版本为0.8
首先来看看kafka的架构图 详细了解请参考官方



我这边有三台机器用于kafka 日志收集的

A 192.168.1.1 为server

B 192.168.1.2 为producer

C 192.168.1.3 为consumer
首先在A上的kafka安装目录下执行如下命令

./kafka-server-start.sh ../config/server.properties

启动kafka 通过netstat -npl 可以查看出是否开启默认端口9092
B为我们的nginx日志产生服务器，在这里的日志是网站实时写入到access-nginx.log 中

因此我们可以通过 tail -f 的方式能看到当前网站正在请求的日志信息。如果你的网站访问量很大请勿执行tail -f
同样我们也要在B上部署kafka,如果你没有写kafka 的客户端的话（查看客户端API地址 ）
执行如下命令来push 数据到集群中

tail -n 0 -f   /www/nh-nginx02/access.log  | bin/kafka-console-producer.sh --broker-list 192.168.1.1:9092 --topic sb-nginx03

这样我们就将日志push到kafka消息中了
C中，现在我们来写 consumer pull数据，还是要部署一下kafka 然后执行命令

bin/kafka-console-consumer.sh --zookeeper 192.168.1.1:2181 --topic sb-nginx03 --from-beginning

参数

–zookeeper 指定了你集群中zookeeper 的地址和端口即可

–topic 要和我们在B中push的时候指定的名称一致
上述方式只为在shell 命令行下，如何通过spark来写consumer呢？

假设你已经下载好spark1.0 源码 假设你已经部署好sbt scala等环境
scala 代码如下：

package test

import java.util.Properties

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf

object KafkaTest {

def main(args:Array[String])
{
if (args.length < 5) {
System.err.println("Usage: KafkaTest <zkQuorum> <group> <topics> <numThreads> <output>")
System.exit(1)
}
val Array(zkQuorum, group, topics, numThreads,output) = args
val sparkConf = new SparkConf().setAppName("KafkaTest")
val ssc =  new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint("checkpoint")

val topicpMap = topics.split(",").map((_,numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicpMap).map(_._2)
lines.saveAsTextFiles(output)
ssc.start()
ssc.awaitTermination()

//.saveAsTextFile(output)

}

}

然后编译

mvn -Phadoop-2.3 -Dhadoop.version=2.3.0-cdh5.0.1 -DskipTests package

然后spark作业提交

./bin/spark-submit  --master local[*]  --class org.apache.spark.fccs.KafkaTest ./test/target/scala-2.10/spark-test-1.0.0-hadoop2.3.0-cdh5.0.1.jar  zoo02 my-test  sb-nginx03 1 hdfs://192.168.1.1:9100/tmp/spark-log.txt

结果如下：



spark scala