Structured streaming

Structured streaming是spark 2.0以后新增的用于实时处理的技术。与spark streaming不同的是,Structured streaming打开了数据源到数据落地之间的限制,它这两个端整合起来,形成真正的“流”,形成一张巨大的表。同时也正因为此特点,真正实现了exactly once语义。

传统的spark streaming处理流程

在spark streaming中可能实现从数据源到计算的"exactly once",但在数据落地的时候,并不能。比如,数据处理完毕,写入redis的过程中,集群崩掉。那么重启后,这部份数据会丢掉或者重复消费。除非自己去实现。而在Structured streaming中都已经得到了较好的实现。

Structured streaming处理流程

话不多说,先来个官网的例子,最直观的感受

 1 import org.apache.spark.sql.SparkSession
 2 import org.apache.spark.sql.streaming.ProcessingTime
 3 
 4 object structured{
 5 
 6   def main(args: Array[String]) {
 7     val spark = SparkSession
 8       .builder
 9       .appName("StructuredNetworkWordCount")
10       .master("local")
11       .getOrCreate()
12 
13     import spark.implicits._
14 
15     val ds1 = spark
16       .readStream
17       .format("kafka")
18       .option("kafka.bootstrap.servers", "master:9092")
19       .option("subscribe", "test2")
20       .load()
21 
22     System.setProperty("hadoop.home.dir", "\hadoop-common-2.2.0-bin-master")
23     spark.sparkContext.setCheckpointDir("/chekpoint")
24 
25     val ds2 = ds1.selectExpr("CAST (value as STRING) ").as[String]
26 
27     val words = ds2.as[String].flatMap(_.split(" "))
28 
29     val wordCounts = words.groupBy("value").count()
30 
31     val query = words
32       .writeStream
33       .outputMode("append")
34       .format("console")
35       .trigger(ProcessingTime("11 seconds"))
36       .start()
37 
38     query.awaitTermination()
39   }
40 
41 }
View Code

最直观的可以看到:

1:跟spark streaming不同,Structured streaming已经纳入了spark sql的框架

2:统一了api,以kafka作为消息源为例,原来的KakfaUtils.Dsteam或才directDstream统一变成了read,readStream,输出统一为write,writeStream

3:没有了sparksteramingcontext,计算的启动和等待,任务执行间隔时间设置都变成了由数据启动

需要注意的地方

1:.master("local")

2:System.setProperty("hadoop.home.dir", "e:\hadoop-common-2.2.0-bin-master")  如果是在windows本地测试,需要设置hadoop本地环境

3:spark.sparkContext.setCheckpointDir("e:/chekpoint")  Structured streaming 目前(2.1.1)支持三种输出模式,除了append以外都需要checkpoint的支持,可是append不支持聚合操作(Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets),使用场景实在有限

4:ds1.selectExpr 一定要使用一个dataset去接收

5:定时设置 .trigger(ProcessingTime("10 seconds")) 或者 .trigger(ProcessingTime(10,TimeUnit.SECONDS))

问题

1:没有了sparksteramingcontext,一些sparkconf参数怎么设置。比如说,控制kafka消费速率,

     .set("spark.streaming.backpressure.enabled", "true")
     .set("spark.streaming.receiver.maxRate", "3")
     .set("spark.streaming.kafka.maxRatePerPartition", "12")

由spark.sparkContext.getConf.set(key, value)设置?效果如何?有无其它方法?

2:统一了读取消息的api以后,就kafka而言,还有没有低阶和高阶api之分(DStream和DirectDStream)?是否需要自己自己去控制offset?

3:输出模式的异同。append不允许聚合操作,那么使用场景非常有限,意义何在?complete会保留完整的数据(以往各批次),在窗口操作方面较以前的话更有优势,但是时间长了会不会占用非常多的资源?它的内部机制?如何只想得到当次数据,但又要做聚合操作,有没有方法?update是最新(2.1.1)才正式上线的功能。

4:实现exactly once依靠的是 .foreach(new ForeachWriter[Row]一条一条处理来实现的?那么,可以批量写提升效率吗?

 未完待续

原文地址:https://www.cnblogs.com/eryuan/p/7001894.html