在Spark Stream中创建DataFrame
在Spark Stream中创建DataFrame
提问于 2017-07-10 13:35:13
我已经把Kafka Stream和Spark连接起来了此外，我还训练了Apache Spark Mlib模型，以基于流文本进行预测。我的问题是，要得到一个预测，我需要通过一个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 streaming不允许创建DataFrame。有没有办法做到这一点？我可以使用case类或struct吗？
回答 1
Stack Overflow用户
回答已采纳
发布于 2017-07-10 16:59:51
可以从RDD创建DataFrame或Dataset，就像在core Spark中一样。为此，我们需要应用一个模式。然后，在foreachRDD中，我们可以将生成的RDD转换为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
}页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/45004411
复制相关文章
相似问题
腾讯云开发者
