Sof*_*ker 4 apache-spark apache-zeppelin
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.eventhubs.EventHubsUtils
import sqlContext.implicits._
val ehParams = Map[String, String](
"eventhubs.policyname" -> "Full",
...
)
val ssc = new StreamingContext(sc, Seconds(2))
val stream = EventHubsUtils.createUnionStream(ssc, ehParams)
val cr = stream.window(Seconds(6))
case class Message(msg: String)
stream.map(msg=>Message(new String(msg))).foreachRDD(rdd=>rdd.toDF().registerTempTable("temp"))
stream.print
ssc.start
Run Code Online (Sandbox Code Playgroud)
以上这个开始并运行良好,但我似乎无法阻止它.对%sql show tables的任何调用都将冻结.
我如何停止上面的StreamingContext?
ssc.stop 还会杀死Spark Context,需要重启解释器.
请ssc.stop(stopSparkContext=false, stopGracefully=true)改用.
| 归档时间: |
|
| 查看次数: |
2659 次 |
| 最近记录: |