Spark Streaming结合Flume、Kafka最新最全日志分析

Spark Streaming结合Flume、Kafka最新最全日志分析

1.修改相应的配置文件

将flume/conf/producer.conf将需要监控的日志输出文件修改为本地的log 路径:
/var/log/nginx/www.eric.aysaas.com-access.log

2.导入相关 jar 包

3.新建log/KafkaTest.scala 代码如下

[code=language-scala]import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils

/**
* flume+kafka+SparkStreaming 实时 nginx 日志获取
* Created by eric on 16/6/29.
*/
object KafkaLog {

def main(agrs: Array[String]): Unit = {

val sparkConf = new SparkConf().setMaster("local[1]").setAppName("StreamingTest")
val ssc = new StreamingContext(sparkConf, Seconds(20))//代表一个给定的秒数的实例

val topic = "HappyBirthDayToAnYuan"
val topicSet = topic.split(" ").toSet

//用 brokers and topics 创建 direct kafka stream
val kafkaParams = Map[String, String]("metadata.broker.list" -> "localhost:9092")

//直接从 kafka brokers 拉取信息,而不使用任何接收器.
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicSet
)
val lines = messages.map(_._2)
lines.print()
val words: DStream[String] = lines.flatMap(_.split("\n"))
words.count().print()

//启动
ssc.start()
ssc.awaitTermination()
}
}

4.Test