在Spark Stream中创建一个DataFrame [英] Create a DataFrame in Spark Stream
问题描述
我已将Kafka Stream连接到Spark.我已经训练了Apache Spark Mlib模型,使其能够基于流文本进行预测.我的问题是,得到一个预测,我需要通过一个DataFramework.
I've connected the Kafka Stream to the Spark. As well as I've trained Apache Spark Mlib model to prediction based on a streamed text. My problem is, get a prediction I need to pass a 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.有什么办法吗?我可以使用案例类或结构吗?
My problem is, Spark streaming doesn't allow to create a DataFrame. Is there any way to do that? Can I use case class or struct?
推荐答案
可以像在核心Spark中一样从RDD创建DataFrame
或Dataset
.为此,我们需要应用一个模式.然后,在foreachRDD
内,我们可以将生成的RDD转换为DataFrame,该数据帧可以进一步与ML管道一起使用.
It's possible to create a DataFrame
or Dataset
from an RDD as you would in core Spark. To do that, we need to apply a schema. Within the foreachRDD
we can then transform the resulting RDD into a DataFrame that can be further used with an ML pipeline.
// 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
}
这篇关于在Spark Stream中创建一个DataFrame的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!