本文主要向大家介绍了【云计算】StructuredStreaming消费kafka的数据案例，通过具体的内容向大家展现，希望对大家学习云计算有所帮助。

最近也是有很多同学问我,StructuredStreaming结合kafka的使用,我简单的写了一个demo,后续会有更加具体和详细的介绍,今天先来一个简单的demo吧.代码测试过了,可以运行.

package spark

import org.apache.spark.sql.{SparkSession}

/**
  * structredstreaming
  */
object StructuredStreaming {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder.appName("StructuredStreaming").getOrCreate()
    val df = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers","")
    .option("subscribe","jason_0606")
    .option("startingOffsets", "latest")
    .load()
    import  spark.implicits._
    val query = df.select($"key", $"value")
      .as[(String, String)].map(kv => kv._1 + " " + kv._2).as[String]
      .writeStream
      .outputMode("append")
      .format("console")
      .start()
    println(query +"哈哈哈")
    query.awaitTermination()
  }
}

今天就先写到这里,有时间继续更新.

本文由职坐标整理并发布，希望对同学们有所帮助。了解更多详情请关注职坐标大数据云计算大数据安全频道！