Scala を使用して Spark スタンドアロン クラスターで実行する Kafka ストリーム プログラムを作成しました。コードは私のローカルで正常に動作します。Azure VM で Kafka 、Cassandra、および Spark のセットアップを行いました。ポートのブロックを回避するために、すべての受信ポートと送信ポートを開きました。
マスター始めました
sbin>./start-master.sh
開始スレーブ
sbin# ./start-slave.sh spark://vm-hostname:7077
マスター WEB UI でこのステータスを確認しました。
ジョブを送信
bin#./spark-submit --class xyStreamJob --master spark://vm-hostname:7077 /home/user/appl.jar
アプリケーションが追加され、Master WEB UI に表示されていることに気付きました。
トピックにいくつかのメッセージを発行しましたが、メッセージが受信されず、Cassandra DB に保持されます。
マスター Web コンソールでアプリケーション名をクリックすると、そのアプリケーション コンソール ページで [ストリーミング] タブが利用できないことに気付きました。
アプリケーションが VM で動作せず、ローカルで正常に動作するのはなぜですか?
VM で問題をデバッグする方法は?
def main(args: Array[String]): Unit = {
val spark = SparkHelper.getOrCreateSparkSession()
val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
spark.sparkContext.setLogLevel("WARN")
val kafkaStream = {
val kafkaParams = Map[String, Object](
"bootstrap.servers" ->
"vmip:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "loc",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val topics = Array("hello")
val numPartitionsOfInputTopic = 3
val streams = (1 to numPartitionsOfInputTopic) map {
_ => KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) )
}
streams
}
kafkaStream.foreach(rdd=> {
rdd.foreachRDD(conRec=> {
val offsetRanges = conRec.asInstanceOf[HasOffsetRanges].offsetRanges
conRec.foreach(str=> {
try {
println(str.value().trim)
CassandraHelper.saveItemEvent(str.value().trim)
}catch {
case ex: Exception => {
println(ex.getMessage)
}
}
})
rdd.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
})
println("Read Msg")
})
println(" Spark parallel reader is ready !!!")
ssc.start()
ssc.awaitTermination()
}
def getSparkConf(): SparkConf = {
val conf = new SparkConf(true)
.setAppName("TestAppl")
.set("spark.cassandra.connection.host", "vmip")
.set("spark.streaming.stopGracefullyOnShutdown","true")
.setMaster("spark://vm-hostname:7077")
conf
}
バージョン
scalaVersion := "2.11.8"
val sparkVersion = "2.2.0"
val connectorVersion = "2.0.7"
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-core" % sparkVersion %"provided",
"org.apache.spark" %% "spark-sql" % sparkVersion %"provided",
"org.apache.spark" %% "spark-hive" % sparkVersion %"provided",
"com.datastax.spark" %% "spark-cassandra-connector" % connectorVersion ,
"org.apache.kafka" %% "kafka" % "0.10.1.0",
"org.apache.spark" %% "spark-streaming-kafka-0-10" % sparkVersion,
"org.apache.spark" %% "spark-streaming" % sparkVersion %"provided",
)
mergeStrategy in assembly := {
case PathList("org", "apache", "spark", "unused", "UnusedStubClass.class") => MergeStrategy.first
case x => (mergeStrategy in assembly).value(x)
}