Structed Streaming 小案例

1 首先是官网：

http://spark.apache.org/docs/latest/structured-streaming-programming-guide.ht

2.注意官方文档中的着重表示的地方例如（黑体加重，斜体等）
我们都知道spark streaming 是基于spark core API
那Structed Streaming基于的是什么？

        没错就是Spark SQL。 所以Data /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的引入，总之和f 相互发展促进。

Finally:最后贴一个小的 案例：

      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_ (col("value"), path = "$.uri").alias("uri"),      get_json_ (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()}