Искровой поток не выполняет иск SQL запрос

Question

Я столкнулся с одной проблемой при выполнении spark sql поверх потоковой обработки spark.

Я не печатаю значение x в строке var x = sqlContext.sql ("select count (*) fromЦены ")

, пожалуйста, найдите мой код здесь

    import spark.implicits._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Encoders
import org.apache.spark.streaming._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import spark.implicits._
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import spark.implicits._
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.storage.StorageLevel
import java.util.regex.Pattern
import java.util.regex.Matcher
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql._

val conf = new SparkConf().setAppName("streamHive").setMaster("local[*]").set("spark.driver.allowMultipleContexts", "true")
val ssc = new StreamingContext(conf, Seconds(5))    
val sc=ssc.sparkContext

val lines = ssc.textFileStream("file:///home/sdf/testHive")
case class Prices(name: String, age: String,sex: String, location: String)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

def parse (rdd : org.apache.spark.rdd.RDD[String] ) = {
var l = rdd.map(_.split(","))
val prices = l.map(p => Prices(p(0),p(1),p(2),p(3)))
val pricesDf = sqlContext.createDataFrame(prices)
pricesDf.registerTempTable("prices")
println("showing printdfShow")
pricesDf.show()
var x = sqlContext.sql("select count(*) from prices")
println("hello")
println (x)
}

lines.foreachRDD { rdd => parse(rdd)}

ssc.start()

Я получаю следующий результат, это не печатает искра SQL результат:

   [count(1): bigint]
   showing printdfShow
   +----+---+---+--------+
   |name|age|sex|location|
   +----+---+---+--------+
   +----+---+---+--------+

   hello
   [count(1): bigint]
   showing printdfShow
   +----+---+---+--------+
   |name|age|sex|location|
   +----+---+---+--------+
   | rop| 22|  M|      uk|
   | fop| 24|  F|      us|
   | dop| 23|  M|     fok|
   +----+---+---+--------+

   hello
   [count(1): bigint]
   showing printdfShow
   +----+---+---+--------+
   |name|age|sex|location|
   +----+---+---+--------+
   +----+---+---+--------+

   hello
   [count(1): bigint]

пожалуйста, помогите мне,как использовать spark sql в потоковом искре, так как я новичок в spark.

Answer 1

Пожалуйста, попробуйте это в своем коде после ценDf.show

println(pricesDf.count)

Если вы хотите это в том же коде, попробуйте ниже, а не println (x)

x.show

xэто фрейм данных, а не значение, поэтому он не печатается во время выполнения println (x).Чтобы получить его в переменной, вы можете попробовать это

println(x.rdd.map(r => r.getString(0)).collect()(0))