这是在Spark Streaming上运行简单SQL查询的代码。
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.Duration
object StreamingSQL {
case class Persons(name: String, age: Int)
def main(args: Array[String]) {
val sparkConf = new SparkConf().setMaster("local").setAppName("HdfsWordCount")
val sc = new SparkContext(sparkConf)
// Create the context
val ssc = new StreamingContext(sc, Seconds(2))
val lines = ssc.textFileStream("C:/Users/pravesh.jain/Desktop/people/")
lines.foreachRDD(rdd=>rdd.foreach(println))
val sqc = new SQLContext(sc);
import sqc.createSchemaRDD
// Create the FileInputDStream on the directory and use the
// stream to count words in new files created
lines.foreachRDD(rdd=>{
rdd.map(_.split(",")).map(p => Persons(p(0), p(1).trim.toInt)).registerAsTable("data")
val teenagers = sqc.sql("SELECT name FROM data WHERE age >= 13 AND age <= 19")
teenagers.foreach(println)
})
ssc.start()
ssc.awaitTermination()
}
}
正如您所看到的,要在流上运行SQL,必须在foreachRDD方法内部进行查询。我想对从两个不同流接收的数据运行SQL联接。有什么办法可以做到吗?
好吧,我想总结一下我们在Spiro的回答中讨论后得出的解决方法。他建议先创建一个空表,然后在其中插入RDD,这一建议很成功。唯一的问题是Spark还不允许插入表。以下是可以做的:
首先,创建一个RDD,该RDD具有与您期望的流模式相同的模式:
import sqlContext.createSchemaRDD
val d1=sc.parallelize(Array(("a",10),("b",3))).map(e=>Rec(e._1,e._2))
然后将其保存为Parquet文件
d1.saveAsParquetFile("/home/p1.parquet")
现在,加载镶木地板文件,并使用registerAsTable()方法将其注册为表。
val parquetFile = sqlContext.parquetFile("/home/p1.parquet")
parquetFile.registerAsTable("data")
现在,当您收到流时,只需在流上应用foreachRDD(),并使用insertInto()法在上面创建的表中继续插入各个RDD
dStream.foreachRDD(rdd=>{
rdd.insertInto("data")
})
此insertInto()工作正常,并允许将数据收集到表中。现在,您可以对任意数量的流执行同样的操作,然后运行查询。
按照编写代码的方式,每次运行SQL查询时都会生成一系列小SchemaRDD。诀窍是将它们中的每一个保存到累加RDD或累加表中。
第一个,使用insertInto
:的表格方法
对于每个流,首先创建一个emty RDD,将其注册为表,获得一个空表。举个例子,假设你称之为"所有青少年"。
然后,对于每个查询,使用SchemaRDD的insertInto
方法将结果添加到该表中:
teenagers.insertInto("allTeenagers")
如果您对两个流都这样做,创建两个单独的累积表,那么您可以使用一个普通的旧SQL查询来连接它们。
(注意:我实际上还没能让他工作,一点点搜索让我怀疑其他人是否已经工作了,但我很确定我已经理解insertInto
的设计意图,所以我认为这个解决方案值得记录。)
第二个,unionAll
方法(也有union
方法,但这使正确获取类型变得更加困难):
这涉及到创建一个初始RDD——再次称之为allTeenagers
。
// create initial SchemaRDD even if it's empty, so the types work out right
var allTeenagers = sqc.sql("SELECT ...")
然后,每次:
val teenagers = sqc.sql("SELECT ...")
allTeenagers = allTeenagers.unionAll(teenagers)
也许不用说,您需要列进行匹配。