Structured Streaming 实战案例 读取Scoker

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

object WordCount {
  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")
    //2.接收数据
    val dataDF: DataFrame = spark.readStream
      .option("host", "node01")
      .option("port", 9999)
      .format("socket")
      .load()
    //3.处理数据
    import spark.implicits._
    val dataDS: Dataset[String] = dataDF.as[String]
    val wordDS: Dataset[String] = dataDS.flatMap(_.split(" "))
    val result: Dataset[Row] = wordDS.groupBy("value").count().sort($"count".desc)
    //result.show()
    //Queries with streaming sources must be executed with writeStream.start();
    result.writeStream
      .format("console")//往控制台写
      .outputMode("complete")//每次将所有的数据写出
      .trigger(Trigger.ProcessingTime(0))//触发时间间隔,0表示尽可能的快
      .option("checkpointLocation","./810")//设置checkpoint目录,用来做合并
      .start()//开启
      .awaitTermination()//等待停止
  }
}

import org.apache.spark.SparkContext

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

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

object WordCount {

  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")

    //2.接收数据

    val dataDF: DataFrame = spark.readStream

      .option("host", "node01")

      .option("port", 9999)

      .format("socket")

      .load()

    //3.处理数据

    import spark.implicits._

    val dataDS: Dataset[String] = dataDF.as[String]

    val wordDS: Dataset[String] = dataDS.flatMap(_.split(" "))

    val result: Dataset[Row] = wordDS.groupBy("value").count().sort($"count".desc)

    //result.show()

    //Queries with streaming sources must be executed with writeStream.start();

    result.writeStream

      .format("console")//往控制台写

      .outputMode("complete")//每次将所有的数据写出

      .trigger(Trigger.ProcessingTime(0))//触发时间间隔,0表示尽可能的快

      .option("checkpointLocation","./810")//设置checkpoint目录,用来做合并

      .start()//开启

      .awaitTermination()//等待停止

  }

}