4 
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

上記は起動して正常に動作しますが、停止できないようです。%sql show tables を呼び出すと、フリーズするだけです。

上記の StreamingContext を停止するにはどうすればよいですか?

4

2 に答える 2

6

ssc.stopまた、Spark コンテキストを強制終了し、インタープリターの再起動が必要になります。

ssc.stop(stopSparkContext=false, stopGracefully=true)代わりに使用してください。

于 2016-07-14T00:38:14.917 に答える
1

ssc.stop新しい段落でそれを停止する必要があります

ストリーミング プラットフォームとの統合を改善する方法については、dev@ メーリング リストでも進行中の議論があります。

于 2015-09-07T02:58:14.497 に答える