spark与kafka连接测试

叁歲伎倆 2022-07-12 23:28 134阅读 0赞

测试环境：scala版本为2.11.8,jdk版本为java1.7.79  
搭建的工程为maven工程，所需要的依赖有：

<dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming_2.11</artifactId>
                <version>2.0.1</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
                <version>2.0.1</version>
            </dependency>

scala代码有：

package sparkStreaming__
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}
    import org.apache.spark.streaming.kafka.KafkaUtils
    
    /** * Created by xiaopengpeng on 2016/12/12. * */
    /*class Spark_kafka_123 { }*/
    object Spark_kafka_123{
      def main(args: Array[String]): Unit = {
        if (args.length!=4){
          System.err.println("Usage:Spark_kafka<zkQuorum><group><topic><numThreads>")
          System.exit(1)
        }
        val Array(zkQuorum,group,topic,numThreads) = args
        val conf = new SparkConf().setAppName("Spark_kafka").setMaster("local[*]")
        val ssc = new StreamingContext(conf,Seconds(5))
        ssc.checkpoint("checkpoint")
        val topicMap = topic.split(",").map((_,numThreads.toInt)).toMap
        val lines = KafkaUtils.createStream(ssc,zkQuorum,group,topicMap).map(_._2)
    
        val words = lines.flatMap(_.split(" "))
        val wordCounts = words.map(x=>(x,1L)).reduceByKeyAndWindow(_+_,_-_,Minutes(10),Seconds(20),5)
        wordCounts.print
        ssc.start()
        ssc.awaitTermination()
      }
    }

发表评论取消回复

表情：

评论列表（有 0 条评论，134人围观）

还没有评论，来说两句吧...

相关阅读

相关 spark与kafka连接测试

测试环境：scala版本为2.11.8,jdk版本为java1.7.79 搭建的工程为maven工程，所需要的依赖有： <dependency>

叁歲伎倆/ 2022年07月12日 23:28/ 0 赞/ 135 阅读

相关 Spark streaming kafka OffsetOutOfRangeException 异常分析与解决

[Spark streaming kafka OffsetOutOfRangeException 异常分析与解决][Spark streaming kafka OffsetOu

桃扇骨/ 2022年06月18日 03:19/ 0 赞/ 183 阅读

相关 Spark 与 Kafka 集成出错: Apache Spark: java.lang.NoSuchMethodError

按照 spark 与 kafka 的教程一步一步进行操作, 在运行 KafkaWordCount 的示例时，却始终没有预期的输出. 如果正确的话，大概是这样: ...

偏执的太偏执、/ 2022年06月17日 12:44/ 0 赞/ 137 阅读

相关 Kafka总结（九）：KafKa 与Spark整合应用

[Kafka总结（一）：Kafka概述][Kafka_Kafka] [Kafka总结（二）：Kafka核心组件][Kafka_Kafka 1] [Kafka总结（三）：Ka

冷不防/ 2022年05月18日 04:41/ 0 赞/ 435 阅读

相关 spark+kafka+hive

使用spark streaming消费kafka消息，并通过hql同步到hive中用到的pom： <!-- https://mvnrepository.com/a

ゞ浴缸里的玫瑰/ 2022年05月14日 07:21/ 0 赞/ 275 阅读

相关 spark消费kafka消息

教程： [http://spark.apache.org/docs/2.2.0/streaming-kafka-0-10-integration.html][http_spa

待我称王封你为后i/ 2022年05月14日 05:45/ 0 赞/ 294 阅读

相关六.Spark Streaming之kafka容错性测试

第一步.部署和启动三个broker: 不了解如何启动的参考[链接][Link 1] 第二步.测试 1).通过命令查看节点的详细信息: kafka-to

电玩女神/ 2022年04月15日 00:46/ 0 赞/ 135 阅读

相关 kafka和spark总结

kafka和spark总结本文涉及到的技术版本号： scala 2.11.8 kafka1.1.0 spark2.3.1 kafka简介

迈不过友情╰/ 2022年03月28日 20:38/ 0 赞/ 262 阅读

相关 Spark学习（7）- Spark Streaming整合Kafka

流程： 1. 创建maven项目 2. 引入相应的依赖 3. 编写代码 4. 发布第一种方式：KafkaUtils.createDstream方式（基于kafka

桃扇骨/ 2022年03月10日 03:42/ 0 赞/ 230 阅读

相关 kafka+spark(待整理)

import org.apache.kafka.common.serialization.StringDeserializer import org.apach

电玩女神/ 2021年11月26日 13:14/ 0 赞/ 293 阅读