标签：String val kafka 整合 import apache org SparkStreaming Kafka

intellij代码实现Spark Stream和Kafka结合消息接收及处理StreamKaf

StreamKafkaProducer1

package org.hnsw

import java.util

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamKafkaProducer1 {

  def streamingkafka() = {
    val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[4]")
//      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    val streamRdd = ssc.socketTextStream("192.168.3.66",8888)
    // 分割字符
    val words = streamRdd.flatMap((x)=>{
      x.split(" ")
    }).map((x)=>{
      (x,1)
    })
    // 第一步 统计每个单词数目
    val wordscount = words.reduceByKeyAndWindow((v1:Int,v2:Int)=>{
      v1+v2
    },Seconds(30),Seconds(10))
    wordscount.print()
    //第二步 发送消息到kafka
    wordscount.foreachRDD((rdd)=>{
      rdd.foreachPartition((partRdd)=>{
        //设置kafka服务参数
        val props = new util.HashMap[String,Object]()
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.3.66:9092")
        //序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer])
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer])
        //实例化一个kafka生产者
        val producer = new KafkaProducer[String,String](props)  //创建kafka生产者
        //封装成kafka消息
        //定义topic 及消息内容
        val topic="kafka-lt" //指定发送的topic名字
        partRdd.foreach((line)=>{
          val str = line._1 +" "+line._2 //定义发送单词结构为 "单词 次数"
          val message = new ProducerRecord[String,String](topic,null,str) //封装kafka消息
          //给kafka发消息
          producer.send(message) //发送消息
        })
      })
    })
    ssc.start()
    ssc.awaitTermination()
  }

  def main(args: Array[String]): Unit = {
    streamingkafka()
  }
}

标签：String,val,kafka,整合,import,apache,org,SparkStreaming,Kafka
来源： https://www.cnblogs.com/857weir09432/p/16362819.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。