企业网站设计注意事项代写文章质量高的平台
目录
0. 相关文章链接
1. 基本操作
1.1. 弱类型 api
1.2. 强类型
1.3. 直接执行 sql
2. 基于 event-time 的窗口操作
2.1. event-time 窗口理解
2.2. event-time 窗口生成规则
3. 基于 Watermark 处理延迟数据
3.1. 什么是 Watermark 机制
3.2. update 模式下使用 watermark
3.3. append 模式下使用 wartermark
3.4. watermark 机制总结
4. 流数据去重
5. join操作
5.1. Stream-static Joins
5.1.1. 内连接
5.1.2. 外连接
5.2. Stream-stream Joins
5.2.1. inner join
4.2.2. outer join
6. Streaming DF/DS 不支持的操作
0. 相关文章链接
Spark文章汇总
1. 基本操作
在 DF/DS 上大多数通用操作都支持作用在 Streaming DataFrame/Streaming DataSet 上。
准备处理数据： people.json
{"name": "Michael","age": 29,"sex": "female"}
{"name": "Andy","age": 30,"sex": "male"}
{"name": "Justin","age": 19,"sex": "male"}
{"name": "Lisi","age": 18,"sex": "male"}
{"name": "zs","age": 10,"sex": "female"}
{"name": "zhiling","age": 40,"sex": "female"}1.1. 弱类型 api
代码示例：
import org.apache.spark.sql.types.{LongType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}object StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 创建格式，并读取数据val peopleSchema: StructType = new StructType().add("name", StringType).add("age", LongType).add("sex", StringType)val peopleDF: DataFrame = spark.readStream.schema(peopleSchema).json("/Project/Data/json")// 弱类型 apival df: DataFrame = peopleDF.select("name", "age", "sex").where("age > 20")df.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}结果输出：
-------------------------------------------
Batch: 0
-------------------------------------------
+-------+---+------+
| name|age| sex|
+-------+---+------+
|Michael| 29|female|
| Andy| 30| male|
|zhiling| 40|female|
+-------+---+------+1.2. 强类型
代码示例：
import org.apache.spark.sql.types.{LongType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}object StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 创建格式，并读取数据val peopleSchema: StructType = new StructType().add("name", StringType).add("age", LongType).add("sex", StringType)val peopleDF: DataFrame = spark.readStream.schema(peopleSchema).json("/Project/Data/json")// 强类型，转成 dsval peopleDS: Dataset[People] = peopleDF.as[People]val df: Dataset[String] = peopleDS.filter((_: People).age > 20).map((_: People).name)df.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}case class People(name: String, age: Long, sex: String)结果输出：
-------------------------------------------
Batch: 0
-------------------------------------------
+-------+
| value|
+-------+
|Michael|
| Andy|
|zhiling|
+-------+1.3. 直接执行 sql
代码示例：
import org.apache.spark.sql.types.{LongType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}object StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 创建格式，并读取数据val peopleSchema: StructType = new StructType().add("name", StringType).add("age", LongType).add("sex", StringType)val peopleDF: DataFrame = spark.readStream.schema(peopleSchema).json("/Project/Data/json")// 直接执行SQL，创建临时表peopleDF.createOrReplaceTempView("people")val df: DataFrame = spark.sql("select * from people where age > 20")df.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}结果输出：
-------------------------------------------
Batch: 0
-------------------------------------------
+-------+---+------+
| name|age| sex|
+-------+---+------+
|Michael| 29|female|
| Andy| 30| male|
|zhiling| 40|female|
+-------+---+------+2. 基于 event-time 的窗口操作
2.1. event-time 窗口理解
在 Structured Streaming 中, 可以按照事件发生时的时间对数据进行聚合操作, 即基于 event-time 进行操作。在这种机制下, 即不必考虑 Spark 陆续接收事件的顺序是否与事件发生的顺序一致, 也不必考虑事件到达 Spark 的时间与事件发生时间的关系。因此, 它在提高数据处理精度的同时, 大大减少了开发者的工作量。我们现在想计算 10 分钟内的单词, 每 5 分钟更新一次, 也就是说在 10 分钟窗口 12:00 - 12:10, 12:05 - 12:15, 12:10 - 12:20等之间收到的单词量。 注意, 12:00 - 12:10 表示数据在 12:00 之后 12:10 之前到达。现在,考虑一下在 12:07 收到的单词。单词应该增加对应于两个窗口12:00 - 12:10和12:05 - 12:15的计数。因此,计数将由分组键（即单词）和窗口（可以从事件时间计算）索引。
统计后的结果应该是这样的：
代码示例：
import org.apache.spark.sql.functions.window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 使用socket数据源val lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).option("includeTimestamp", value = true) // 给产生的数据自动添加时间戳.load// 把行切割成单词, 保留时间戳val words: DataFrame = lines.as[(String, Timestamp)].flatMap((line: (String, Timestamp)) => {line._1.split(" ").map(((_: String), line._2))}).toDF("word", "timestamp")// 按照窗口和单词分组, 并且计算每组的单词的个数，最后按照窗口排序val wordCounts: Dataset[Row] = words.groupBy(// 调用 window 函数, 返回的是一个 Column 类型// 参数 1: df 中表示时间戳的列// 参数 2: 窗口长度// 参数 3: 滑动步长window($"timestamp", "60 seconds", "10 seconds"),$"word").count().orderBy($"window")wordCounts.writeStream.outputMode("complete").format("console").option("truncate", "false") // 不截断.为了在控制台能看到完整信息, 最好设置为 false.start.awaitTermination()// 关闭执行环境spark.stop()}
}结果输出：
-------------------------------------------
Batch: 0
-------------------------------------------
+------+----+-----+
|window|word|count|
+------+----+-----+
+------+----+-----+-------------------------------------------
Batch: 1
-------------------------------------------
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-02 21:18:00, 2023-08-02 21:19:00]|abc |3 |
|[2023-08-02 21:18:10, 2023-08-02 21:19:10]|a |3 |
|[2023-08-02 21:18:10, 2023-08-02 21:19:10]|abc |5 |
|[2023-08-02 21:18:20, 2023-08-02 21:19:20]|abc |5 |
|[2023-08-02 21:18:20, 2023-08-02 21:19:20]|a |3 |
|[2023-08-02 21:18:30, 2023-08-02 21:19:30]|a |3 |
|[2023-08-02 21:18:30, 2023-08-02 21:19:30]|abc |5 |
|[2023-08-02 21:18:40, 2023-08-02 21:19:40]|abc |5 |
|[2023-08-02 21:18:40, 2023-08-02 21:19:40]|a |3 |
|[2023-08-02 21:18:50, 2023-08-02 21:19:50]|a |3 |
|[2023-08-02 21:18:50, 2023-08-02 21:19:50]|abc |5 |
|[2023-08-02 21:19:00, 2023-08-02 21:20:00]|a |3 |
|[2023-08-02 21:19:00, 2023-08-02 21:20:00]|abc |2 |
+------------------------------------------+----+-----+-------------------------------------------
Batch: 2
-------------------------------------------
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-02 21:18:00, 2023-08-02 21:19:00]|abc |3 |
|[2023-08-02 21:18:10, 2023-08-02 21:19:10]|a |3 |
|[2023-08-02 21:18:10, 2023-08-02 21:19:10]|abc |5 |
|[2023-08-02 21:18:20, 2023-08-02 21:19:20]|abc |5 |
|[2023-08-02 21:18:20, 2023-08-02 21:19:20]|a |3 |
|[2023-08-02 21:18:30, 2023-08-02 21:19:30]|a |3 |
|[2023-08-02 21:18:30, 2023-08-02 21:19:30]|abc |5 |
|[2023-08-02 21:18:40, 2023-08-02 21:19:40]|abc |5 |
|[2023-08-02 21:18:40, 2023-08-02 21:19:40]|a |3 |
|[2023-08-02 21:18:50, 2023-08-02 21:19:50]|a |3 |
|[2023-08-02 21:18:50, 2023-08-02 21:19:50]|abc |5 |
|[2023-08-02 21:19:00, 2023-08-02 21:20:00]|a |3 |
|[2023-08-02 21:19:00, 2023-08-02 21:20:00]|abc |2 |
|[2023-08-02 21:19:10, 2023-08-02 21:20:10]|a |2 |
|[2023-08-02 21:19:10, 2023-08-02 21:20:10]|abc |1 |
|[2023-08-02 21:19:20, 2023-08-02 21:20:20]|a |2 |
|[2023-08-02 21:19:20, 2023-08-02 21:20:20]|abc |1 |
|[2023-08-02 21:19:30, 2023-08-02 21:20:30]|a |2 |
|[2023-08-02 21:19:30, 2023-08-02 21:20:30]|abc |1 |
|[2023-08-02 21:19:40, 2023-08-02 21:20:40]|a |2 |
+------------------------------------------+----+-----+
only showing top 20 rows由此可以看出, 在这种窗口机制下, 无论事件何时到达, 以怎样的顺序到达, Structured Streaming 总会根据事件时间生成对应的若干个时间窗口, 然后按照指定的规则聚合。
2.2. event-time 窗口生成规则
可以查看 org.apache.spark.sql.catalyst.analysis.TimeWindowing 类下的如下代码：
The windows are calculated as below:
maxNumOverlapping <- ceil(windowDuration / slideDuration)
for (i <- 0 until maxNumOverlapping)windowId <- ceil((timestamp - startTime) / slideDuration)windowStart <- windowId * slideDuration + (i - maxNumOverlapping) * slideDuration + startTimewindowEnd <- windowStart + windowDurationreturn windowStart, windowEnd将event-time 作为“初始窗口”的结束时间, 然后按照窗口滑动宽度逐渐向时间轴前方推进, 直到某个窗口不再包含该 event-time 为止。 最终以“初始窗口”与“结束窗口”之间的若干个窗口作为最终生成的 event-time 的时间窗口。
每个窗口的起始时间与结束时间都是前必后开的区间, 因此初始窗口和结束窗口都不会包含 event-time, 最终不会被使用。
得到窗口如下：
3. 基于 Watermark 处理延迟数据
3.1. 什么是 Watermark 机制
在数据分析系统中, Structured Streaming 可以持续的按照 event-time 聚合数据, 然而在此过程中并不能保证数据按照时间的先后依次到达。 例如: 当前接收的某一条数据的 event-time 可能远远早于之前已经处理过的 event-time。 在发生这种情况时, 往往需要结合业务需求对延迟数据进行过滤。现在考虑如果事件延迟到达会有哪些影响。 假如, 一个单词在 12:04(event-time) 产生, 在 12:11 到达应用。 应用应该使用 12:04 来在窗口(12:00 - 12:10)中更新计数, 而不是使用 12:11。 这些情况在我们基于窗口的聚合中是自然发生的, 因为结构化流可以长时间维持部分聚合的中间状态。
但是, 如果这个查询运行数天, 系统很有必要限制内存中累积的中间状态的数量。 这意味着系统需要知道何时从内存状态中删除旧聚合, 因为应用不再接受该聚合的后期数据。为了实现这个需求, 从 spark2.1, 引入了 watermark(水印), 使用引擎可以自动的跟踪当前的事件时间, 并据此尝试删除旧状态。通过指定 event-time 列和预估事件的延迟时间上限来定义一个查询的 watermark。 针对一个以时间 T 结束的窗口, 引擎会保留状态和允许延迟时间直到(max event time seen by the engine - late threshold > T)。 换句话说, 延迟时间在上限内的被聚合, 延迟时间超出上限的开始被丢弃。
可以通过withWatermark() 来定义watermark，watermark 计算方式：watermark = MaxEventTime - Threshhod；而且, watermark只能逐渐增加, 不能减少。
Structured Streaming 引入 Watermark 机制, 主要是为了解决以下两个问题:
- 处理聚合中的延迟数据
- 减少内存中维护的聚合状态.
注意：在不同输出模式(complete, append, update)中, Watermark 会产生不同的影响。
3.2. update 模式下使用 watermark
在 update 模式下, 仅输出与之前批次的结果相比, 涉及更新或新增的数据。
代码示例如下：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 使用socket数据源val lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load// 输入的数据中包含时间戳, 而不是自动添加的时间戳val words: DataFrame = lines.as[String].flatMap((line: String) => {val split: Array[String] = line.split(",")split(1).split(" ").map(((_: String), Timestamp.valueOf(split(0))))}).toDF("word", "timestamp")// 使用 withWatermark 方法，添加watermark, 参数 1: event-time 所在列的列名 参数 2: 延迟时间的上限.val wordCounts: Dataset[Row] = words.withWatermark("timestamp", "2 minutes").groupBy(window($"timestamp", "10 minutes", "2 minutes"), $"word").count()// 数据输出val query: StreamingQuery = wordCounts.writeStream.outputMode("update").trigger(Trigger.ProcessingTime(1000)).format("console").option("truncate", "false").startquery.awaitTermination()// 关闭执行环境spark.stop()}
}初始化的wartmark是 0，通过如下输入的几条数据，可以看到水位线的变化。
第一次输入数据： 2023-08-07 10:55:00,dog 。这个条数据作为第一批数据。 按照window($"timestamp", "10 minutes", "2 minutes")得到 5 个窗口。 由于是第一批, 所有的窗口的结束时间都大于 wartermark(0), 所以 5 个窗口都显示，如下所示：
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-07 10:50:00, 2023-08-07 11:00:00]|dog |1 |
|[2023-08-07 10:54:00, 2023-08-07 11:04:00]|dog |1 |
|[2023-08-07 10:48:00, 2023-08-07 10:58:00]|dog |1 |
|[2023-08-07 10:52:00, 2023-08-07 11:02:00]|dog |1 |
|[2023-08-07 10:46:00, 2023-08-07 10:56:00]|dog |1 |
+------------------------------------------+----+-----+然后根据当前批次中最大的 event-time, 计算出来下次使用的 watermark. 本批次只有一个数据(10:55), 所有: watermark = 10:55 - 2min = 10:53 。
第二次输入数据： 2023-08-07 11:00:00,dog 。 这条数据作为第二批数据, 计算得到 5 个窗口。 此时的watermark=10:53, 所有的窗口的结束时间均大于 watermark。 在 update 模式下, 只输出结果表中涉及更新或新增的数据。
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-07 10:58:00, 2023-08-07 11:08:00]|dog |1 |
|[2023-08-07 10:54:00, 2023-08-07 11:04:00]|dog |2 |
|[2023-08-07 10:56:00, 2023-08-07 11:06:00]|dog |1 |
|[2023-08-07 10:52:00, 2023-08-07 11:02:00]|dog |2 |
|[2023-08-07 11:00:00, 2023-08-07 11:10:00]|dog |1 |
+------------------------------------------+----+-----+其中: count 是 2 的表示更新, count 是 1 的表示新增。 没有变化的就没有显示(但是内存中仍然保存着)。此时的的 watermark = 11:00 - 2min = 10:58 。如下数据为在内存中保存着，但是没有打印出来的数据：
|[2023-08-07 10:46:00, 2023-08-07 10:56:00]|dog |1 |
|[2023-08-07 10:48:00, 2023-08-07 10:58:00]|dog |1 |
|[2023-08-07 10:50:00, 2023-08-07 11:00:00]|dog |1 |第三次输入数据： 2023-08-07 10:55:00,dog 。 这条数据作为第 3 批次，相当于一条延迟数据，计算得到 5 个窗口。此时的 watermark = 10:58 当前内存中有两个窗口的结束时间已经低于 10: 58。
|[2023-08-07 10:46:00, 2023-08-07 10:56:00]|dog |1 |
|[2023-08-07 10:48:00, 2023-08-07 10:58:00]|dog |1 |则立即删除这两个窗口在内存中的维护状态。 同时, 当前批次中新加入的数据所划分出来的窗口, 如果窗口结束时间低于 11:58, 则窗口会被过滤掉。
所以这次输出结果：
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-07 10:50:00, 2023-08-07 11:00:00]|dog |2 |
|[2023-08-07 10:54:00, 2023-08-07 11:04:00]|dog |3 |
|[2023-08-07 10:52:00, 2023-08-07 11:02:00]|dog |3 |
+------------------------------------------+----+-----+第三个批次的数据处理完成后, 立即计算: watermark= 10:55 - 2min = 10:53, 这个值小于当前的 watermask(10:58), 所以保持不变(因为 watermask 只能增加不能减少)。
3.3. append 模式下使用 wartermark
代码示例如下：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 使用socket数据源val lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load// 输入的数据中包含时间戳, 而不是自动添加的时间戳val words: DataFrame = lines.as[String].flatMap((line: String) => {val split: Array[String] = line.split(",")split(1).split(" ").map(((_: String), Timestamp.valueOf(split(0))))}).toDF("word", "timestamp")// 使用 withWatermark 方法，添加watermark, 参数 1: event-time 所在列的列名 参数 2: 延迟时间的上限.val wordCounts: Dataset[Row] = words.withWatermark("timestamp", "2 minutes").groupBy(window($"timestamp", "10 minutes", "2 minutes"), $"word").count()// 数据输出val query: StreamingQuery = wordCounts.writeStream.outputMode("append").trigger(Trigger.ProcessingTime(0)).format("console").option("truncate", "false").startquery.awaitTermination()// 关闭执行环境spark.stop()}
}
在 append 模式中, 仅输出新增的数据, 且输出后的数据无法变更。
第一次输入数据： 2023-08-07 10:55:00,dog 。 这个条数据作为第一批数据。 按照window($"timestamp", "10 minutes", "2 minutes")得到 5 个窗口。 由于此时初始 watermask=0, 当前批次中所有窗口的结束时间均大于 watermask。但是 Structured Streaming 无法确定后续批次的数据中是否会更新当前批次的内容。 因此, 基于 Append 模式的特点, 这时并不会输出任何数据(因为输出后数据就无法更改了), 直到某个窗口的结束时间小于 watermask, 即可以确定后续数据不会再变更该窗口的聚合结果时才会将其输出, 并移除内存中对应窗口的聚合状态。
+------+----+-----+
|window|word|count|
+------+----+-----+
+------+----+-----+然后根据当前批次中最大的 event-time, 计算出来下次使用的 watermark。 本批次只有一个数据(10:55), 所有: watermark = 10:55 - 2min = 10:53
第二次输入数据： 2023-08-07 11:00:00,dog 。这条数据作为第二批数据, 计算得到 5 个窗口。 此时的watermark=10:53, 所有的窗口的结束时间均大于 watermark, 仍然不会输出。
+------+----+-----+
|window|word|count|
+------+----+-----+
+------+----+-----+然后计算 watermark = 11:00 - 2min = 10:58
第三次输入数据： 2023-08-07 10:55:00,dog 。相当于一条延迟数据,这条数据作为第 3 批次, 计算得到 5 个窗口。 此时的 watermark = 10:58 当前内存中有两个窗口的结束时间已经低于 10: 58。
|[2023-08-07 10:48:00, 2023-08-07 10:58:00]|dog |1 |
|[2023-08-07 10:46:00, 2023-08-07 10:56:00]|dog |1 |则意味着这两个窗口的数据不会再发生变化, 此时输出这个两个窗口的聚合结果, 并在内存中清除这两个窗口的状态。所以这次输出结果:
+------------------------------------------+----+-----+
|window |word|count|
+------------------------------------------+----+-----+
|[2023-08-07 10:48:00, 2023-08-07 10:58:00]|dog |1 |
|[2023-08-07 10:46:00, 2023-08-07 10:56:00]|dog |1 |
+------------------------------------------+----+-----+第三个批次的数据处理完成后, 立即计算: watermark= 10:55 - 2min = 10:53, 这个值小于当前的 watermask(10:58), 所以保持不变。(因为 watermask 只能增加不能减少)
3.4. watermark 机制总结
- watermark 在用于基于时间的状态聚合操作时, 该时间可以基于窗口, 也可以基于 event-timeb本身。
- 输出模式必须是append或update。 在输出模式是complete的时候(必须有聚合), 要求每次输出所有的聚合结果。 我们使用 watermark 的目的是丢弃一些过时聚合数据, 所以complete模式使用wartermark无效也无意义。
- 在输出模式是append时, 必须设置 watermask 才能使用聚合操作。 其实, watermask 定义了 append 模式中何时输出聚合聚合结果(状态), 并清理过期状态。
- 在输出模式是update时, watermask 主要用于过滤过期数据并及时清理过期状态。
- watermask 会在处理当前批次数据时更新, 并且会在处理下一个批次数据时生效使用。 但如果节点发送故障, 则可能延迟若干批次生效。
- withWatermark 必须使用与聚合操作中的时间戳列是同一列。df.withWatermark("time", "1 min").groupBy("time2").count() 无效。
- withWatermark 必须在聚合之前调用。 f.groupBy("time").count().withWatermark("time", "1 min") 无效。
4. 流数据去重
需求内容：根据唯一的 id 实现数据去重
代码示例：
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 使用socket数据源val lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load// 数据预处理val words: DataFrame = lines.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), Timestamp.valueOf(arr(1)), arr(2))}).toDF("uid", "ts", "word")// 去重重复数据 uid 相同就是重复. 可以传递多个列val wordCounts: Dataset[Row] = words.withWatermark("ts", "2 minutes").dropDuplicates("uid")// 输出数据wordCounts.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}数据输入（按顺序从上到下）：
1,2023-08-09 11:50:00,dog
2,2023-08-09 11:51:00,dog
1,2023-08-09 11:50:00,dog
3,2023-08-09 11:53:00,dog
1,2023-08-09 11:50:00,dog
4,2023-08-09 11:45:00,dog注意点：
- dropDuplicates 不可用在聚合之后, 即通过聚合得到的 df/ds 不能调用dropDuplicates
- 使用watermask - 如果重复记录的到达时间有上限,则可以在事件时间列上定义水印,并使用guid和事件时间列进行重复数据删除。该查询将使用水印从过去的记录中删除旧的状态数据,这些记录不会再被重复。这限制了查询必须维护的状态量。
- 没有watermask - 由于重复记录可能到达时没有界限,查询将来自所有过去记录的数据存储为状态。
测试：
- 第一次输入数据：1,2023-08-09 11:50:00,dog
+---+-------------------+----+
|uid| ts|word|
+---+-------------------+----+
| 1|2023-08-09 11:50:00| dog|
+---+-------------------+----+- 第二次输入数据：2,2023-08-09 11:51:00,dog
+---+-------------------+----+
|uid| ts|word|
+---+-------------------+----+
| 2|2023-08-09 11:51:00| dog|
+---+-------------------+----+- 第三次输入数据：1,2023-08-09 11:50:00,dog （id 重复无输出）
- 第四次输入数据：3,2023-08-09 11:53:00,dog （此时 watermask=11:51）
+---+-------------------+----+
|uid| ts|word|
+---+-------------------+----+
| 3|2023-08-09 11:53:00| dog|
+---+-------------------+----+- 第五次输入数据：1,2023-08-09 11:50:00,dog （数据重复, 并且数据过期, 所以无输出）
- 第六次输入数据：4,2023-08-09 11:45:00,dog （数据过时, 所以无输出）
5. join操作
Structured Streaming 支持 streaming DataSet/DataFrame 与静态的DataSet/DataFrame 进行 join, 也支持 streaming DataSet/DataFrame与另外一个streaming DataSet/DataFrame 进行 join。join 的结果也是持续不断的生成, 类似于前面的 streaming 的聚合结果。
5.1. Stream-static Joins
静态数据：
lisi,male
zhiling,female
zs,male流式数据：
lisi,20
zhiling,40
ww,305.1.1. 内连接
代码示例：
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}object StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 1. 静态 dfval arr: Array[(String, String)] = Array(("lisi", "male"), ("zhiling", "female"), ("zs", "male"));val staticDF: DataFrame = spark.sparkContext.parallelize(arr).toDF("name", "sex")// 2. 流式 dfval lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()val streamDF: DataFrame = lines.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1).toInt)}).toDF("name", "age")// 3. join 等值内连接 a.name=b.nameval joinResult: DataFrame = streamDF.join(staticDF, "name")// 4. 输出joinResult.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}数据输出：
+-------+---+------+
| name|age| sex|
+-------+---+------+
|zhiling| 40|female|
| lisi| 20| male|
+-------+---+------+5.1.2. 外连接
代码示例：
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}object StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 1. 静态 dfval arr: Array[(String, String)] = Array(("lisi", "male"), ("zhiling", "female"), ("zs", "male"));val staticDF: DataFrame = spark.sparkContext.parallelize(arr).toDF("name", "sex")// 2. 流式 dfval lines: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()val streamDF: DataFrame = lines.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1).toInt)}).toDF("name", "age")// 3. joinval joinResult: DataFrame = streamDF.join(staticDF, Seq("name"), "left")// 4. 输出joinResult.writeStream.outputMode("append").format("console").start.awaitTermination()// 关闭执行环境spark.stop()}
}数据输出：
+-------+---+------+
| name|age| sex|
+-------+---+------+
|zhiling| 40|female|
| ww| 30| null|
| lisi| 20| male|
+-------+---+------+5.2. Stream-stream Joins
在 Spark2.3, 开始支持 stream-stream join。Spark 会自动维护两个流的状态, 以保障后续流入的数据能够和之前流入的数据发生 join 操作, 但这会导致状态无限增长。 因此, 在对两个流进行 join 操作时, 依然可以用 watermark 机制来消除过期的状态, 避免状态无限增长。
第 1 个数据格式：姓名,年龄,事件时间
lisi,female,2023-08-09 11:50:00
zs,male,2023-08-09 11:51:00
ww,female,2023-08-09 11:52:00
zhiling,female,2023-08-09 11:53:00
fengjie,female,2023-08-09 11:54:00
yifei,female,2023-08-09 11:55:00第 2 个数据格式：姓名,年龄,事件时间
lisi,18,2023-08-09 11:50:00
zs,19,2023-08-09 11:51:00
ww,20,2023-08-09 11:52:00
zhiling,22,2023-08-09 11:53:00
yifei,30,2023-08-09 11:54:00
fengjie,98,2023-08-09 11:55:005.2.1. inner join
对 2 个流式数据进行 join 操作，输出模式仅支持append模式。
不带 watermast 的 inner join（join 的速度很慢）：
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 第 1 个 streamval nameSexStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1), Timestamp.valueOf(arr(2)))}).toDF("name", "sex", "ts1")// 第 2 个 streamval nameAgeStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 8888).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1).toInt, Timestamp.valueOf(arr(2)))}).toDF("name", "age", "ts2")// join 操作val joinResult: DataFrame = nameSexStream.join(nameAgeStream, "name")// 数据输出joinResult.writeStream.outputMode("append").format("console").trigger(Trigger.ProcessingTime(0)).start().awaitTermination()// 关闭执行环境spark.stop()}
}// 数据输出：
// +-------+------+-------------------+---+-------------------+
// | name| sex| ts1|age| ts2|
// +-------+------+-------------------+---+-------------------+
// |zhiling|female|2023-08-09 11:53:00| 22|2023-08-09 11:53:00|
// | ww|female|2023-08-09 11:52:00| 20|2023-08-09 11:52:00|
// | yifei|female|2023-08-09 11:55:00| 30|2023-08-09 11:54:00|
// | zs| male|2023-08-09 11:51:00| 19|2023-08-09 11:51:00|
// |fengjie|female|2023-08-09 11:54:00| 98|2023-08-09 11:55:00|
// | lisi|female|2023-08-09 11:50:00| 18|2023-08-09 11:50:00|
// +-------+------+-------------------+---+-------------------+
带 watermast 的 inner join（join 的速度很慢）：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 第 1 个 streamval nameSexStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1), Timestamp.valueOf(arr(2)))}).toDF("name1", "sex", "ts1")// 第 2 个 streamval nameAgeStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 8888).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1).toInt, Timestamp.valueOf(arr(2)))}).toDF("name2", "age", "ts2").withWatermark("ts2", "1 minutes")// join 操作val joinResult: DataFrame = nameSexStream.join(nameAgeStream,expr("""|name1=name2 and|ts2 >= ts1 and|ts2 <= ts1 + interval 1 minutes""".stripMargin))// 数据输出joinResult.writeStream.outputMode("append").format("console").trigger(Trigger.ProcessingTime(0)).start().awaitTermination()// 关闭执行环境spark.stop()}
}// 数据输出：
// +-------+------+-------------------+-------+---+-------------------+
// | name1| sex| ts1| name2|age| ts2|
// +-------+------+-------------------+-------+---+-------------------+
// |zhiling|female|2023-08-09 11:53:00|zhiling| 22|2023-08-09 11:53:00|
// | ww|female|2023-08-09 11:52:00| ww| 20|2023-08-09 11:52:00|
// | zs| male|2023-08-09 11:51:00| zs| 19|2023-08-09 11:51:00|
// |fengjie|female|2023-08-09 11:54:00|fengjie| 98|2023-08-09 11:55:00|
// | lisi|female|2023-08-09 11:50:00| lisi| 18|2023-08-09 11:50:00|
// +-------+------+-------------------+-------+---+-------------------+4.2.2. outer join
外连接必须使用 watermast，和内连接相比, 代码几乎一致, 只需要在连接的时候指定下连接类型即可：joinType = "left"。
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}import java.sql.Timestampobject StreamTest {def main(args: Array[String]): Unit = {// 创建 SparkSession. 因为 ss 是基于 spark sql 引擎, 所以需要先创建 SparkSessionval spark: SparkSession = SparkSession.builder().master("local[*]").appName("StreamTest").getOrCreate()import spark.implicits._// 第 1 个 streamval nameSexStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1), Timestamp.valueOf(arr(2)))}).toDF("name1", "sex", "ts1")// 第 2 个 streamval nameAgeStream: DataFrame = spark.readStream.format("socket").option("host", "localhost").option("port", 8888).load.as[String].map((line: String) => {val arr: Array[String] = line.split(",")(arr(0), arr(1).toInt, Timestamp.valueOf(arr(2)))}).toDF("name2", "age", "ts2").withWatermark("ts2", "1 minutes")// join 操作val joinResult: DataFrame = nameSexStream.join(nameAgeStream,expr("""|name1=name2 and|ts2 >= ts1 and|ts2 <= ts1 + interval 1 minutes""".stripMargin),joinType = "left")// 数据输出joinResult.writeStream.outputMode("append").format("console").trigger(Trigger.ProcessingTime(0)).start().awaitTermination()// 关闭执行环境spark.stop()}
}// 数据输出：
// +-------+------+-------------------+-------+---+-------------------+
// | name1| sex| ts1| name2|age| ts2|
// +-------+------+-------------------+-------+---+-------------------+
// |zhiling|female|2023-08-09 11:53:00|zhiling| 22|2023-08-09 11:53:00|
// | ww|female|2023-08-09 11:52:00| ww| 20|2023-08-09 11:52:00|
// | zs| male|2023-08-09 11:51:00| zs| 19|2023-08-09 11:51:00|
// |fengjie|female|2023-08-09 11:54:00|fengjie| 98|2023-08-09 11:55:00|
// | lisi|female|2023-08-09 11:50:00| lisi| 18|2023-08-09 11:50:00|
// +-------+------+-------------------+-------+---+-------------------+
6. Streaming DF/DS 不支持的操作
到目前, DF/DS 的有些操作 Streaming DF/DS 还不支持：
- 多个Streaming 聚合(例如在 DF 上的聚合链)目前还不支持
- limit 和取前 N 行还不支持
- distinct 也不支持
- 仅仅支持对 complete 模式下的聚合操作进行排序操作
- 仅支持有限的外连接
- 有些方法不能直接用于查询和返回结果, 因为他们用在流式数据上没有意义
- count() 不能返回单行数据, 必须是s.groupBy().count()
- foreach() 不能直接使用, 而是使用: ds.writeStream.foreach(...)
- show() 不能直接使用, 而是使用 console sink
如果执行上面操作会看到这样的异常: operation XYZ is not supported with streaming DataFrames/Datasets。
注：其他Spark相关系列文章链接由此进 -> Spark文章汇总