在dstream.transform（）中使用SQL而不是Spark流？

Question

在foreachrdd()中有一些使用SQL over Spark Streaming的示例。但是如果我想在tranform()中使用SQL:

case class AlertMsg(host:String, count:Int, sum:Double)
val lines = ssc.socketTextStream("localhost", 8888)
lines.transform( rdd => {
  if (rdd.count > 0) {
    val t = sqc.jsonRDD(rdd)
    t.registerTempTable("logstash")
    val sqlreport = sqc.sql("SELECT host, COUNT(host) AS host_c, AVG(lineno) AS line_a FROM logstash WHERE path = '/var/log/system.log' AND lineno > 70 GROUP BY host ORDER BY host_c DESC LIMIT 100")
    sqlreport.map(r => AlertMsg(r(0).toString,r(1).toString.toInt,r(2).toString.toDouble))
  } else {
    rdd
  }
}).print()

我犯了这样的错误：

[error]/users/raochenlin/downloads/spark-1.2.0-bin-hadoop2.4/logstash/src/main/scala/logstash.scala:52:没有html" target="_blank">方法转换的类型参数：（transformfunc:org.apache.spark.rdd.rdd.rdd[String]=>org.apache.spark.rdd.rdd[U])（隐式证据$5:scala.reflect.classtag[U])org.apache.spark.rdd.dstream.dstream[U]，因此可以将其应用于参数(org.apache.spark.rdd.rdd[String]=>org.apache.spark.rdd.rdd[String]字符串<:java.io.serializable])[error]----因为--[error]参数表达式的类型与形参类型不兼容；[error]找到：org.apache.spark.rdd.rdd[String]=>org.apache.spark.rdd.rdd[_>:logstash.alertmsg带有字符串<:java.io.serializable][err]必需：org.apache.spark.rdd.rdd[String]=>org.apache.spark.rdd.rdd[？u][err]lines.transform（rdd=>{[errer]^[err]找到一个错误[errer]（compile：compile）编译失败

似乎只有使用sqlreport.map(r=>r.tostring)才是正确的用法？

Answer 1

dstream.transform接受一个函数transformfunc:(rdd[T])¥rdd[U]在本例中，if必须在两个条件的计算中产生相同的类型，但情况并非如此：

if (count == 0) => RDD[String]
if (count > 0) => RDD[AlertMsg]

在本例中，如果rdd.count...，则删除的优化，以便具有唯一的转换路径。