最近也是有很多同學問我,StructuredStreaming結合kafka的使用,我簡單的寫了一個wordcount的demo,後續會有更加具體和詳細的介紹,今天先來一個簡單的demo吧.程式碼測試過了,可以執行.

package spark

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

/**
  * structredstreaming消費kafka的資料,實現exactly-once的語義;
  */
object StructuredStreaming {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.INFO)
    Logger.getLogger("org.apache.kafka.clients.consumer").setLevel(Level.ERROR)
    var query:org.apache.spark.sql.streaming.StreamingQuery = null
    val spark = SparkSession.builder.appName("StructuredStreaming").getOrCreate()
    val topic = "jason_1027"
    val broker = "master:9092,storm1:9092,storm2:9092"
    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers",broker)
      .option("subscribe",topic)
      .option("startingOffsets", "latest")
      .option("max.poll.records", 10000)
      .load()
    import  spark.implicits._
    val word = df.selectExpr("CAST(value AS STRING)").as[String].flatMap(_.split(" "))
    val wordcount = word.groupBy("value").count()
    val q = wordcount
      .writeStream
      .outputMode("complete")
      .format("console")
      .start()
    q.awaitTermination()
  }
}
 

今天就先寫到這裡,有時間繼續更新.

如果有寫的不對的地方,歡迎大家指正,如果有什麼疑問,可以加QQ群:340297350,謝謝