如何将kafka流转换为spark rdd或sparkDataframe

7rfyedvj 于 2021-06-07 发布在 Kafka

关注(0)|答案(2)|浏览(471)

我试着从Kafka加载数据，这是成功的，但我无法转换为Sparkrdd，

val kafkaParams = Map("metadata.broker.list" -> "IP:6667,IP:6667")
val offsetRanges = Array(
    OffsetRange("first_topic", 0,1,1000)
  )
val ssc = new StreamingContext(new SparkConf, Seconds(60))
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

现在我如何读取这个流对象？？？我的意思是把它转换成sparkDataframe并执行一些计算
我试着转换成Dataframe

stream.foreachRDD { rdd =>
     println("Hello")
      import sqlContext.implicits._
      val dataFrame = rdd.map {case (key, value) => Row(key, value)}.toDf()
    }

但是todf不是工作错误：值todf不是org.apache.spark.rdd.rdd[org.apache.spark.sql.row]的成员

scala apache-kafka apache-spark spark-streaming

来源：https://stackoverflow.com/questions/35177920/how-to-convert-kafka-stream-to-spark-rdd-or-spark-dataframe

2条答案

按热度按时间

hmmo2u0o1#

它很旧，但我认为您在从行创建df时忘记了添加模式：

val df =  sc.parallelize(List(1,2,3)).toDF("a")
val someRDD = df.rdd
val newDF = spark.createDataFrame(someRDD, df.schema)

（在Spark壳2.2.0中测试）

赞(0）回复(0）举报 2021-06-07

yyyllmsg2#

val kafkaParams = Map("metadata.broker.list" -> "IP:6667,IP:6667")
val offsetRanges = Array(
    OffsetRange("first_topic", 0,1,1000)
  )
val ssc = new StreamingContext(new SparkConf, Seconds(60))
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) 

val lines = stream.map(_.value)
val words = lines.flatMap(_.split(" ")).print()   //def createDataFrame(words: RDD[Row], Schema: StructType)

// Start your computation then
ssc.start()
ssc.awaitTermination()

赞(0）回复(0）举报 2021-06-07

我来回答

如何将kafka流转换为spark rdd或sparkDataframe

2条答案

相关问题

热门标签

最新问答