在spark流中创建Dataframe

a6b3iqyw 于 2021-06-08 发布在 Kafka

关注(0)|答案(1)|浏览(445)

我已经把Kafka河和Spark连接起来了。我还训练了apachesparkmlib模型来基于流式文本进行预测。我的问题是，得到一个预测，我需要传递一个dataframework。

//kafka stream    
val stream = KafkaUtils.createDirectStream[String, String](
          ssc,
          PreferConsistent,
          Subscribe[String, String](topics, kafkaParams)
        )
//load mlib model
val model = PipelineModel.load(modelPath)
 stream.foreachRDD { rdd =>

      rdd.foreach { record =>
       //to get a prediction need to pass DF
       val toPredict = spark.createDataFrame(Seq(
          (1L, record.value())
        )).toDF("id", "review")
        val prediction = model.transform(test)
      }
}

我的问题是，spark流不允许创建Dataframe。有什么办法吗？我可以使用case类或结构吗？

apache-kafka apache-spark apache-spark-mllib spark-streaming sparse-matrix

来源：https://stackoverflow.com/questions/45004411/create-a-dataframe-in-spark-stream

1条答案

按热度按时间

nafvub8i1#

可以创建一个 DataFrame 或者 Dataset 就像你在核心星火里一样。为此，我们需要应用一个模式。在 foreachRDD 然后，我们可以将得到的rdd转换成一个Dataframe，该Dataframe可以进一步与ml管道一起使用。

// we use a schema in the form of a case class
case class MyStructure(field:type, ....)
// and we implement our custom transformation from string to our structure
object MyStructure {
    def parse(str: String) : Option[MyStructure] = ...
}

val stream = KafkaUtils.createDirectStream... 
// give the stream a schema using a case class
val strucStream =  stream.flatMap(cr => MyStructure.parse(cr.value))

strucStream.foreachRDD { rdd =>
    import sparkSession.implicits._
    val df = rdd.toDF()
    val prediction = model.transform(df)
    // do something with df
}

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

我来回答

在spark流中创建Dataframe

1条答案

相关问题

热门标签

最新问答