Boa noite
Estou fazendo um teste com sparkstreaming consumindo de um tópico do kafka. Quando start o job ele apresenta a seguinte msg.mensagem:
Segue meu código: import org.apache.spark._ import org.apache.spark.streaming._ import org.apache.spark.streaming.kafka010._ import org.apache.kafka.clients.consumer.ConsumerConfig
import scala.collection.mutable
object WorldCount_Kafka { def main(args: Array[String]): Unit = {
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.clients.consumer.ConsumerConfig
import scala.collection.mutable
object WorldCount_Kafka {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local[2]")
.setAppName("WorldCount-kafka")
//Read messages in batch of 20 seconds
val ssc = new StreamingContext(conf, Seconds(20))
val sc = ssc.sparkContext
sc.setLogLevel("ERROR")
val kafkaParam = new mutable.HashMap[String, String]()
kafkaParam.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092")
kafkaParam.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer")
kafkaParam.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer")
kafkaParam.put(ConsumerConfig.GROUP_ID_CONFIG, "test")
kafkaParam.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
kafkaParam.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")
//Configure Spark to listen messages in topic test
val topicList = List("first_topic")
// Read value of each message from Kafka and return it
val messageStream = KafkaUtils.createDirectStream(ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String,String](topicList, kafkaParam)
)
val lines = messageStream.map(consumerRecord => consumerRecord.value().asInstanceOf[String])
// Break every message into words and return list of words
val words = lines.flatMap(_.split(" "))
// Take every word and return Tuple with (word,1)
val wordMap = words.map(word => (word, 1))
// Count occurance of each word
val wordCount = wordMap.reduceByKey((first, second) => first + second)
//Print the word count
wordCount.print()
ssc.start()
ssc.awaitTermination()
}
}
} }