6

我正在尝试对 Spark 中的流数据运行 SQL 查询。这看起来很简单，但是当我尝试它时，我得到错误table not found : tablename >。它无法找到我注册的表。

将 Spark SQL 与批处理数据一起使用可以正常工作，所以我认为这与我调用 streamingcontext.start() 的方式有关。任何想法是什么问题？这是代码：

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.sql.SQLContext

object Streaming {

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 sqc = new SQLContext(sc);
    import sqc.createSchemaRDD

    // Create the FileInputDStream on the directory and use the
    // stream to count words in new files created
    val lines = ssc.textFileStream("C:/Users/pravesh.jain/Desktop/people.txt")
    lines.foreachRDD(rdd=>rdd.map(_.split(",")).map(p => Persons(p(0), p(1).trim.toInt)).registerAsTable("data"))
    // lines.foreachRDD(rdd=>rdd.foreach(println))
    val teenagers = sqc.sql("SELECT name FROM data WHERE age >= 13 AND age <= 19")
    ssc.start()
    ssc.awaitTermination()
  }
}

欢迎任何建议。谢谢。

4

1 回答 1

12

好吧，我知道了问题所在。您必须在 foreachRDD 函数中查询数据，否则无法识别该表。像这样的工作：

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 Mlist {

  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.txt")
    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()
  }
}
于 2014-08-21T06:17:15.520 回答