【Spark Streaming】(四)基于 Spark Structured Streaming 的开发与数据处理
文章目录
- 一、前言
- 二、Spark Streaming vs Structured Streaming
- 2.1 Spark Streaming
- 2.2 Structured Streaming
- 2.3 区别
- 2.3.1 流模型
- 2.3.2 RDD vs DataFrame、DataSet
- 2.3.3 Process Time vs Event Time
- 2.3.4 可靠性保障
- 2.3.5 sink
- 2.4 总结
- 三、Spark Structured Streaming 特性
- 四、Spark Structured Streaming 开发
- 4.1 Handling Event-time
- 4.2 Handling Late Data
- 4.3 Using DataFrame & Dataset
- 4.4 “Legal” Join Types
- 4.5 Streaming Deduplication (流去重)
- 4.6 Unsupported / Changed Operations (不支持/更改的操作)
- 4.7 Input Sources for Structured Streaming (输入源)
一、前言
通过前几篇博客 Spark Streaming 架构及工作原理、DStream 编码实战 ,想必大家都 Spark Streaming 已经有了一定的了解。
这篇博客,我会和大家讲下如何基于 Spark Structured Streaming 进行开发与数据处理 。 首先我们来比较下,Structured Streaming 与 Spark Streaming 它有哪些不同呢? 带着 Structured Streaming 为啥这么牛Σ(っ °Д °;)っ 的疑问,我们开始今天的学习吧!
Tips:我会在两者比较的时候介绍 Spark Structured Streaming 的一些特性,请留意。
二、Spark Streaming vs Structured Streaming
2.1 Spark Streaming
Spark Streaming 是 Spark 最初的流处理框架,使用了微批的形式来进行流处理。
提供了基于 RDDs 的 Dstream API,每个时间间隔内的数据为一个RDD,源源不断对 RDD 进行处理来实现流计算。
2.2 Structured Streaming
Spark 2.X 出来的流框架,采用了无界表的概念,流数据相当于往一个表上不断追加行。
基于 Spark SQL 引擎实现,可以使用大多数 Spark SQL 的 function。
2.3 区别
2.3.1 流模型
Spark Streaming
Spark Streaming 采用微批的处理方法。每一个批处理间隔的为一个批,也就是一个RDD,我们对RDD进行操作就可以源源不断的接收、处理数据。
Structured Streaming
“Input Table” is Unbounded - 无界的输入表
Structured Streaming is to treat a live data stream as a table that is being continuously appended .
Structured Streaming 将实时数据当做被连续追加的表,流上的每一条数据都类似于将一行新数据添加到表中。
以上图为例,每隔1秒从输入源获取数据到 Input Table,并触发Query计算,然后将结果写入 Result Table,之后根据指定的 Output模式(有三种) 进行写出。
- Complete Mode (完整模式)整个更新后的结果将被写入外部存储
- Append Mode (default) (追加模式)由于将写入最后一个触发器,因此结果表中只会追加新行
- Update Mode (更新模式) 只写入结果表中因为最后一个触发器而更新的行
上面的1秒是指定的触发间隔(trigger interval),如果不指定的话,先前数据的处理完成后,系统将立即检查是否有新数据。
With every trigger interval, new rows get appended to the input table.
需要注意的是,Spark Streaming 本身设计就是一批批的以批处理间隔划分RDD;而Structured Streaming中并没有提出批的概念,Structured Streaming 按照每个 Trigger Interval(触发间隔) 接收数据到 Input Table,将数据处理后再追加到无边界的 Result Table 中,想要何种方式输出结果取决于指定的模式。所以,虽说 Structured Streaming 也有类似于 Spark Streaming 的 Interval,其本质概念是不一样的。Structured Streaming更像流模式。
2.3.2 RDD vs DataFrame、DataSet
Spark Streaming中的DStream编程接口是RDD,我们需要对RDD进行处理,处理起来较为费劲且不美观。
stream.foreachRDD(rdd => {balabala(rdd)
})
Structured Streaming使用DataFrame、DataSet的编程接口,处理数据时可以使用Spark SQL中提供的方法,数据的转换和输出会变得更加简单。
spark.readStream.format("kafka").option("kafka.bootstrap.servers", "hadoop01:9092").option("subscribe", "order_data").load().select($"value".cast("string")).as[String].writeStream.outputMode("complete").format("console")
2.3.3 Process Time vs Event Time
Process Time:流处理引擎接收到数据的时间
Event Time:事件真正发生的时间
Spark Streaming中由于其微批的概念,会将一段时间内接收的数据放入一个批内,进而对数据进行处理。划分批的时间是Process Time,而不是Event Time,Spark Streaming没有提供对Event Time的支持。
Structured Streaming提供了基于事件时间处理数据的功能,如果数据包含事件的时间戳,就可以基于事件时间进行处理。
这里以窗口计数为例说明一下区别:
我们这里以10分钟为窗口间隔,5分钟为滑动间隔,每隔5分钟统计过去10分钟网站的pv
假设有一些迟到的点击数据,其本身事件时间是12:01,被spark接收到的时间是12:11;在Spark Streaming的统计中,会毫不犹豫的将它算作是12:05-12:15这个范围内的pv,这显然是不恰当的;在Structured Streaming中,可以使用事件时间将它划分到12:00-12:10的范围内,这才是我们想要的效果。
2.3.4 可靠性保障
两者在可靠性保证方面都是使用了checkpoint机制。
checkpoint通过设置检查点,将数据保存到文件系统,在出现出故障的时候进行数据恢复。
在Spark Streaming中,如果我们需要修改流程序的代码,在修改代码重新提交任务时,是不能从checkpoint中恢复数据的(程序就跑不起来),是因为spark不认识修改后的程序了。
在Structured Streaming中,对于指定的代码修改操作,是不影响修改后从checkpoint中恢复数据的。
2.3.5 sink
二者的输出数据(写入下游)的方式有很大的不同。
Spark Streaming中提供了foreachRDD()
方法,通过自己编程实现将每个批的数据写出。
stream.foreachRDD(rdd => {save(rdd)
})
Structured Streaming 自身提供了一些sink(Console Sink、File Sink、Kafka Sink等),只要通过option配置就可以使用;对于需要自定义的Sink,提供了ForeachWriter的编程接口,实现相关方法就可以完成。
// console sink
val query = res.writeStream.outputMode("append").format("console").start()
2.4 总结
总体来说,Structured Streaming 有更简洁的API、更完善的流功能、更适用于流处理。而spark streaming,更适用于与偏批处理的场景。
三、Spark Structured Streaming 特性
前面已经讲了 Structured Streaming 的基本概念,及其会再讲下它存储、自动流化、容错、性能等方面的特性,在事件时间的处理机制。
因为流处理具有如下显著的复杂性特征,所以很难建立非常健壮的处理过程:
- 一是数据有各种不同格式(Jason、Avro、二进制)、脏数据、不及时且无序;
- 二是复杂的加载过程,基于事件时间的过程需要支持交互查询,和机器学习组合使用;
- 三是不同的存储系统和格式(SQL、NoSQL、Parquet 等),要考虑如何容错。
因为可以运行在Spark SQL引擎上,Spark Structured Streaming天然拥有较好的性能、良好的扩展性及容错性等Spark优势。除此之外,它还具备丰富、统一、高层次的API,因此便于处理复杂的数据和工作流。再加上,无论是Spark自身,还是其集成的多个存储系统,都有丰富的生态圈。这些优势也让Spark Structured Streaming得到更多的发展和使用。
流的定义是一种无限表(unbounded table),把数据流中的新数据追加在这张无限表中,而它的查询过程可以拆解为几个步骤,例如可以从Kafka读取JSON数据,解析JSON数据,存入结构化Parquet表中,并确保端到端的容错机制。其中的特性包括:
- 支持多种消息队列,比如Files/Kafka/Kinesis等。
- 可以用join(), union()连接多个不同类型的数据源。
- 返回一个DataFrame,它具有一个无限表的结构。
- 你可以按需选择SQL(BI分析)、DataFrame(数据科学家分析)、DataSet(数据引擎),它们有几乎一样的语义-和性能。
- 把Kafka的JSON结构的记录转换成String,生成嵌套列,利用了很多优化过的处理函数来完成这个动作,例如from_json(),也允许各种自定义函数协助处理,例如Lambdas, flatMap。
- 在Sink步骤中可以写入外部存储系统,例如Parquet。在Kafka sink中,支持foreach来对输出数据做任何处理,支持事务和exactly-once方式。
- 支持固定时间间隔的微批次处理,具备微批次处理的高性能性,支持低延迟的连续处理(Spark 2.3),支持检查点机制(check point)。
- 秒级处理来自Kafka的结构化源数据,可以充分为查询做好准备。
Spark SQL把批次查询转化为一系列增量执行计划,从而可以分批次地操作数据。
在容错机制上,Structured Streaming 采取检查点机制,把进度offset写入stable的存储中,用JSON的方式保存支持向下兼容,允许从任何错误点(例如自动增加一个过滤来处理中断的数据)进行恢复。这样确保了端到端数据的exactly-once。
在性能上,Structured Streaming 重用了Spark SQL优化器和Tungsten引擎,而且成本降低了3倍!!
Structured Streaming 隔离处理逻辑采用的是可配置化的方式(比如定制JSON的输入数据格式),执行方式是批处理还是流查询很容易识别。这里我们还需要知道批处理、微批次-流处理、持续流处理三种模式的延迟性、吞吐性和资源分配情况。
在时间窗口的支持上,Structured Streaming 支持基于事件时间(event-time)的聚合,这样更容易了解每隔一段时间发生的事情。同时也支持各种用户定义聚合函数(User Defined Aggregate Function,UDAF)。
另外,Structured Streaming可通过不同触发器间分布式存储的状态来进行聚合,状态被存储在内存中,归档采用HDFS的Write Ahead Log (WAL)机制。当然,Structured Streaming还可自动处理过时的数据,更新旧的保存状态。因为历史状态记录可能无限增长,这会带来一些性能问题,为了限制状态记录的大小,Spark使用水印(watermarking)来删除不再更新的旧的聚合数据。允许支持自定义状态函数,比如事件或处理时间的超时,同时支持Scala和Java。
在苹果的信息安全平台中,每秒将产生有百万级事件,Structured Streaming可以用来做缺陷检测,下图是该平台架构:
在该架构中,一是可以把任意原始日志通过ETL加载到结构化日志库中,通过批次控制可很快进行灾难恢复;二是可以连接很多其它的数据信息(DHCP session,缓慢变化的数据);三是提供了多种混合工作方式:实时警告、历史报告、ad-hoc分析、统一的API允许支持各种分析(例如实时报警系统)等,支持快速部署。四是达到了百万事件秒级处理性能。
四、Spark Structured Streaming 开发
4.1 Handling Event-time
Event-time is the time embedded in the data itself, not the time when the data is received. (事件时间是嵌入数据本身的时间,而不是接收数据的时间。)
- The event-time is a column value in the each row (事件时间是每行中的一个列值)
4.2 Handling Late Data
The event-time model allows to handle data that has arrived later than expected. (事件时间模型允许处理晚于预期的数据。)
- Update the old aggregates when there is late data (当有延迟的数据时,更新旧的聚合)
- Clean up old aggregates (清理旧的数据集)
Watermarking(水印)
- A time threshold for how late data can still be handled. (延迟数据处理的时间阈值。)
4.3 Using DataFrame & Dataset
case class WordCount(word: String, count: Int)
val lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()// Split the lines into words
val df = lines.as[String].flatMap(_.split(" ").map(w => (w, 1))).withColumnRenamed("_1", "word").withColumnRenamed("_2", "count")df.createOrReplaceTempView("updates")
val df1 = spark.sql("select count(*) from updates")val df = df.as[WordCount]import org.apache.spark.sql.expressions.scalalang.typed
val result = ds.groupByKey(_.word.substring(0, 1)).agg(typed.sum(_.count))// Start running the query that prints the running counts to the console
val query = result.writeStream.outputMode("complete").format("console").start()
query.awaitTermination()
4.4 “Legal” Join Types
4.5 Streaming Deduplication (流去重)
Duplicate records can be dropped in data streams using a unique identifier in the events. (可以使用事件中的唯一标识符在数据流中删除重复的记录。)
val streamingDf = spark.readStream. ... // columns: guid, eventTime, ...// Without watermark using guid column
streamingDf.dropDuplicates("guid")// With watermark using guid and eventTime columns
streamingDf
.withWatermark("eventTime", "10 seconds")
.dropDuplicates("guid", "eventTime")
4.6 Unsupported / Changed Operations (不支持/更改的操作)
◆ Unsupported Operations:
- Multiple streaming aggregations (i.e. a chain of aggregations on a streaming DF)
- Limit and take first N rows
- Distinct operations on streaming Datasets
- Few types of outer joins on streaming Datasets
- Sorting operations are supported on streaming Datasets only after an aggregation and in Complete Output Mode.
◆ Changed uses:
- count() → ds.groupBy().count
- foreach() → ds.writeStream.foreach( … )
- show() → use the console sink
4.7 Input Sources for Structured Streaming (输入源)
- File source
val eventSchema = new StructType().add("event_id", "string").add("user_id", "string")
.add("start_time", "string").add("address", "string")
.add("city", "string").add("state", "string")
.add("country", "string")
.add("latitude", "float").add("longitude", "float")val dfEvents = spark.readStream.option("sep", ",").schema(eventSchema).csv("hdfs:///user/events/events_*.csv")val result = dfEvents.groupBy("user_id").count()// Start running the query that prints the running counts to the console
val query = result.writeStream.outputMode("complete").format("console").start()query.awaitTermination()
- Kafka source
- Socket source (test only)
- Rate source (test only)
【Spark Streaming】(四)基于 Spark Structured Streaming 的开发与数据处理相关推荐
- Spark(四)— Spark Streaming
Spark(四)- Spark Streaming 一.概述 二.使用 2.1 基础环境 (wordcount测试) 2.2 DStream输出 2.3 自定义采集器 - 对接Kafka 2.4 DS ...
- 【Spark分布式内存计算框架——Structured Streaming】3. Structured Streaming —— 入门案例:WordCount
1.3 入门案例:WordCount 入门案例与SparkStreaming的入门案例基本一致:实时从TCP Socket读取数据(采用nc)实时进行词频统计WordCount,并将结果输出到控制台C ...
- Structured Streaming系列-1、Structured Streaming
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明. 传送门:大数据系列文章目录 官方网址:http://spark.apache.org/. ht ...
- 2021年大数据Spark(四十四):Structured Streaming概述
Apache Spark在2016年的时候启动了Structured Streaming项目,一个基于Spark SQL的全新流计算引擎Structured Streaming,让用户像编写批处理程序 ...
- 大数据Spark(四十四):Structured Streaming概述
文章目录 Structured Streaming概述 Spark Streaming 不足 Structured Streaming 介绍 模块介绍
- 【Spark分布式内存计算框架——Structured Streaming】2. Structured Streaming 核心设计与编程模型
核心设计 2016年,Spark在2.0版本中推出了结构化流处理的模块Structured Streaming,核心设计如下: 第一点:Input and Output(输入和输出) Structur ...
- Spark(四) -- Spark工作机制
版权声明:本文为博主原创文章,未经博主允许不得转载. https://blog.csdn.net/qq1010885678/article/details/45728173 一.应用执行机制 一个应用 ...
- 学习笔记Spark(四)—— Spark编程基础(创建RDD、RDD算子、文件读取与存储)
文章目录 一.创建RDD 1.1.启动Spark shell 1.2.创建RDD 1.2.1.从集合中创建RDD 1.2.2.从外部存储中创建RDD 任务1: 二.RDD算子 2.1.map与flat ...
- 《Spark实战(第2版)》之【Spark 介绍(基于Spark 3.0)】
<Spark实战(第2版)>翻译于新冠疫情在全世界肆虐之时,可谓"生于忧患".短短一年半的时间,全世界数不胜数的人感染了新冠肺炎,甚至死在病毒的攻击之下. ...
- Spark系列四:Spark的经典入门案列之ip地址归属地查询
目录 概述 代码实现 单级模式 分布式模式 方案一: 方案二: 工具类 提示:所有需要的文件我全部放在资源里面了,可以自行下载 概述 需求:根据访问日志的ip地址计算出访问者的归属地,并且按照省份,计 ...
最新文章
- Android Studio 3.5 之后导入第三方Library 库的方法
- Windows10下安装配置 perl 环境
- 噬血代码进不了游戏_玩家认可,二次元魂类游戏,《噬血代码》在三个方面做出了差异化...
- max file descriptors_年轻族的战场!宋MAX强势对比嘉际
- MyBatis框架 动态SQL
- 在python中函数和类都属于可调用对象_在Python中函数和类都属于可调用对象
- python写二分法_Python实现二分法算法实例
- Anaconda安装,linux操作系统学习
- java判断string字符串是不是json格式
- 第三周阶段性小结——Object对象、String类、StringBuffer、StringBuilder、System、Runtime、Date...
- WIN10重新下载安装MicroSoft Store的三种方法
- Win10系统安装教程
- bind 完成正确安装
- csdn前200的大牛
- Python: Command Line Scripts
- C语言中创建自己函数库文件流程
- 2019年个人总结,写在人生不惑之年
- Open VAS 漏扫工具的安装
- 盛志凡:TVOS技术解读
- 计算机专业考编制怎么考,大学这5类专业更适合考编制,有你的专业吗?