Structed Streaming 小案例

1 首先是官網： http://spark.apache.org/docs/latest/structured-streaming-programming-guide.ht

2.注意官方文檔中的着重表示的地方例如（黑體加重，斜體等）

我們都知道spark streaming 是基于spark core API

那Structed Streaming基于的是什麼？

沒錯就是Spark SQL。 是以DataFrame/DataSet API 包括hive的一些functions 不要太好用哦！！！
           

下面這句話：“you can express your streaming computaion the same way

you would express a batch computation on a static data”

就保證了我們在實際開發時的成本比較低，當我們在開發一個的應用中包含流計算和批計算。

3.特點：

可擴充性，容錯性（這都是必備的好嗎？）
精确的一次語義
低延遲
           

4.關鍵點：

Continuously processing
           

databricks的blog上這篇文章寫的也很好

https://databricks.com/blog/2016/07/28/continuous-applications-evolving-streaming-in-apache-spark-2-0.html

5.做到了端到端

延時1ms時能保證至at least one 的語義
延時100ms左右時，能做到 exactly once。
           

6.然後就是流與其他的各種join， watermark的引入，總之和flink 互相發展促進。

Finally:最後貼一個小的 案例：

object testSSApp extends App {

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

  //  結構化流
  private val read = spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("subscribe", "topic1")
    .option("maxOffsetperTrigger", "1000000")
    .option("kafkaConsumer.pollTimeoutMs", "1000")
    .load()

//讀取的kafak 資料為json格式

  val result = read.selectExpr("CAST(value AS STRING)")
    .select(
      get_json_object(col("value"), path = "$.uri").alias("uri"),
      get_json_object(col("value"), path = "$.market").alias("market")
    ).groupBy(window(col("timestamp"),"5min","1min"),
    col("shop"))
    .agg(count("market").alias("uv"),
      approx_count_distinct("uri").alias("pv")).select("*")

  val query = result.writeStream.trigger(Trigger.ProcessingTime(10000)).outputMode("Update")
    .format("console").start()

//這裡展示以console輸出，實際中是回寫到kafak或者外部存儲。

  query.awaitTermination()

}