sparkstreaming之kafka--Spark Streaming对接Kafka的方式一KafkaReceiver
2018-02-18 10:59
651 查看
package com.yys.spark
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Spark Streaming对接Kafka的方式一
*/
object KafkaReceiverWordCount {
def main(args: Array[String]): Unit = {
//使用args传递参数,其中<>表示必须的
if(args.length != 4) {
System.err.println("Usage: KafkaReceiverWordCount <zkQuorum> <group> <topics> <numThreads>")
}
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf() //.setAppName("KafkaReceiverWordCount")
//.setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
// TODO... Spark Streaming如何对接Kafka
val messages = KafkaUtils.createStream(ssc, zkQuorum, group,topicMap)
// TODO... 自己去测试为什么要取第二个
messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
ssc.start()
ssc.awaitTermination()
}
}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* Spark Streaming对接Kafka的方式一
*/
object KafkaReceiverWordCount {
def main(args: Array[String]): Unit = {
//使用args传递参数,其中<>表示必须的
if(args.length != 4) {
System.err.println("Usage: KafkaReceiverWordCount <zkQuorum> <group> <topics> <numThreads>")
}
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf() //.setAppName("KafkaReceiverWordCount")
//.setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
// TODO... Spark Streaming如何对接Kafka
val messages = KafkaUtils.createStream(ssc, zkQuorum, group,topicMap)
// TODO... 自己去测试为什么要取第二个
messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
ssc.start()
ssc.awaitTermination()
}
}
相关文章推荐
- 第90讲,Spark streaming基于kafka 以Receiver方式获取数据 原理和案例实战
- Spark-Streaming获取kafka数据的两种方式:Receiver与Direct的方式
- Spark Streaming之Kafka的Receiver和Direct方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- (转载)Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- sparkstreaming之kafka--Spark Streaming对接Kafka的方式二KafkaDirect
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- Sparkstreaming基于kafka以Receiver方式获取数据原理和案例实战
- Spark-Streaming获取kafka数据的两种方式-Receiver与Direct的方式
- SparkStreaming通过Kafka获取数据(Receiver方式)
- spark streaming 实现kafka的createDirectStream方式!!不坑
- spark-kafka direct方式读取和receiver方式读取的区别
- Kafka到SparkStreaming的两种方式
- Spark Streaming kafka 实现数据零丢失的几种方式
- spark streaming 实现kafka的createDirectStream方式!!不坑