专栏首页笨兔儿Spark2.3.1+Kafka0.9使用Direct模式消费信息异常

Spark2.3.1+Kafka0.9使用Direct模式消费信息异常

Spark2.3.1+Kafka使用Direct模式消费信息

Maven依赖

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

2.3.1spark版本

Direct模式代码

import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Test {

  val zkQuorum = "mirrors.mucang.cn:2181"
  val groupId = "nginx-cg"
  val topic = Map("nginx-log" -> 1)

  val KAFKA_INTERVAL = 10

  case class NginxInof(domain: String, ip: String)

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("NginxLogAnalyze").setMaster("local[*]")
    val sparkContext = new SparkContext(sparkConf)

    val streamContext = new StreamingContext(sparkContext, Seconds(KAFKA_INTERVAL))

    val kafkaParam = Map[String, String](
      "bootstrap.servers" -> "xx.xx.cn:9092",
      "group.id" -> "nginx-cg",
      "auto.offset.reset" -> "largest"
    )

    val topic = Set("nginx-log")

    val kafkaStream = KafkaUtils.createDirectStream(streamContext, kafkaParam, topic)

    val counter = kafkaStream
      .map(_.toString().split(" "))
      .map(item => (item(0).split(",")(1) + "-" + item(2), 1))
      .reduceByKey((x, y) => (x + y))

    counter.foreachRDD(rdd => {
      rdd.foreach(println)
    })


    streamContext.start()
    streamContext.awaitTermination()

  }

}

largest 因为kafka版本过低不支持latest

异常信息

Caused by: java.lang.NoSuchMethodException: scala.runtime.Nothing$.<init>(kafka.utils.VerifiableProperties)
    at java.lang.Class.getConstructor0(Class.java:3082)
    at java.lang.Class.getConstructor(Class.java:1825)
    at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.<init>(KafkaRDD.scala:153)
    at org.apache.spark.streaming.kafka.KafkaRDD.compute(KafkaRDD.scala:136)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    ... 3 more

解决方案

在验证kafka属性时不能使用scala默认的类,需要指定kafka带的类 createDirectStream[String, String, StringDecoder, StringDecoder]其中StringDecoder必须是kafka.serializer.StringDecoder

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • CDH 启用Kerberos安全认证

    看错误信息提示是你没有定义realm,这时需要检查/var/kerberos/krb5kdc/kdc.conf和/etc/krb5kdc.conf两文件是否有定...

    笨兔儿
  • Spark2.3.1使用技巧

    因为在spark-submit时配置的executor-memory 2g等没有生效,后来问同事说他也碰到这样的问题,解决方案就是动态的分配executor,官...

    笨兔儿
  • Idea+maven+scala构建包并在spark on yarn 运行

    在pom.xml配置文件中配置spark开发所需要的包,根据你Spark版本找对应的包,Maven中央仓库

    笨兔儿
  • Spark 3.0.1 Structured Streaming 提交程序异常解决

    先说解决办法,提交时除了添加spark-sql-kafka和kafka-clients jar包外,还要添加spark-token-provider-kafka...

    董可伦
  • org.apache.spark.sql.AnalysisException: Table or view not found: `traintext`.`train`; line 1 pos 14;

    恭喜老铁,跟我遇到了一样的问题,接下来是解决方法: 遇到的问题: org.apache.spark.sql.AnalysisException: Table o...

    用户1171305
  • Spark提交Jar任务警告: Initial job has not accepted any resources;

    WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your...

    时间静止不是简史
  • Spark读取CSV异常 java.lang.ArrayIndexOutOfBoundsException:62

    情况1: 将GBK编码的文件转文UTF-8(我碰见的),当然这种情况也可以用情况2中的解决办法解决~

    董可伦
  • Spark No FileSystem for scheme file 解决方法

    这里的 Local repository 就是项目保存库的位置。在这里面依次打开文件位置:

    机器学习和大数据挖掘
  • spark1.4加载mysql数据 创建Dataframe及join操作连接方法问题

    最后无奈。。就用原来的方法 创建软连接,加载数据,发现可以。。这我就不明白了。。。

    用户3003813
  • spark和kafka jar包冲突NoSuchMethodError: net.jpountz.lz4.LZ4BlockInputStream

    在利用Spark和Kafka处理数据时,有时会同时在maven pom中引入Spark和Kafka的相关依赖。但是当利用Spark SQL处理数据生成的Data...

    大数据学习与分享

扫码关注云+社区

领取腾讯云代金券