Structured Streaming 实战案例 读取文本数据

import org.apache.spark.SparkContext
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
/**
  * {"name":"json","age":23,"hobby":"running"}
  * {"name":"charles","age":32,"hobby":"basketball"}
  * {"name":"tom","age":28,"hobby":"football"}
  * {"name":"lili","age":24,"hobby":"running"}
  * {"name":"bob","age":20,"hobby":"swimming"}
  * 统计年龄小于25岁的人群的爱好排行榜
  */
object WordCount2 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkSession,因为StructuredStreaming的数据模型也是DataFrame/DataSet
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL").getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    val Schema: StructType = new StructType()
      .add("name","string")
      .add("age","integer")
      .add("hobby","string")
    //2.接收数据
    import spark.implicits._
    // Schema must be specified when creating a streaming source DataFrame.
    val dataDF: DataFrame = spark.readStream.schema(Schema).json("D:\\data\\spark\\data")
    //3.处理数据
    val result: Dataset[Row] = dataDF.filter($"age" < 25).groupBy("hobby").count().sort($"count".desc)
    //4.输出结果
    result.writeStream
      .format("console")
      .outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))
      .start()
      .awaitTermination()
  }
}

import org.apache.spark.SparkContext

import org.apache.spark.sql.streaming.Trigger

import org.apache.spark.sql.types.StructType

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**

  * {"name":"json","age":23,"hobby":"running"}

  * {"name":"charles","age":32,"hobby":"basketball"}

  * {"name":"tom","age":28,"hobby":"football"}

  * {"name":"lili","age":24,"hobby":"running"}

  * {"name":"bob","age":20,"hobby":"swimming"}

  * 统计年龄小于25岁的人群的爱好排行榜

  */

object WordCount2 {

  def main(args: Array[String]): Unit = {

    //1.创建SparkSession,因为StructuredStreaming的数据模型也是DataFrame/DataSet

    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL").getOrCreate()

    val sc: SparkContext = spark.sparkContext

    sc.setLogLevel("WARN")

    val Schema: StructType = new StructType()

      .add("name","string")

      .add("age","integer")

      .add("hobby","string")

    //2.接收数据

    import spark.implicits._

    // Schema must be specified when creating a streaming source DataFrame.

    val dataDF: DataFrame = spark.readStream.schema(Schema).json("D:\\data\\spark\\data")

    //3.处理数据

    val result: Dataset[Row] = dataDF.filter($"age" < 25).groupBy("hobby").count().sort($"count".desc)

    //4.输出结果

    result.writeStream

      .format("console")

      .outputMode("complete")

      .trigger(Trigger.ProcessingTime(0))

      .start()

      .awaitTermination()

  }

}