7

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)
}
4

1 に答える 1

0

問題をデバッグするには、まず、メッセージが Kafka を通過することを確認する必要があります。これを行うには、VM でポート 9092 を開き、Kafka から直接使用する必要があります。

bin/kafka-console-consumer.sh --bootstrap-server vmip:9092 --topic hello --from-beginning

from-beginningオプションは、Kafka トピックで構成した最大保持時間まですべてを消費します。

VM に 2 つのバージョンの Spark がないこと、および「spark2-submit」を使用して Spark2 ジョブを送信する必要があることも確認してください。

于 2018-05-07T12:50:45.213 に答える