Spark入门( 八)——Spark流计算新玩法-Structured Streaming

约定不等于承诺〃 2023-02-18 03:24 1阅读 0赞

### Structured Streaming ###

*  简介
 *  快速入门案例
 *   *  程序流程结构
 *  基本概念
 *  故障容错
 *  Structured Streaming API
 *   *  Input Sources
     *   *  File Source
         *   *  Socket source(debug)
         *  Kafka source
     *  Output Sink
     *   *  File sink(Append Mode Only)
         *  KafkaSink((Append|Update|Complete))
         *  Foreach sink(Append|Update|Complate)
         *   *  UserRowWriter
     *  Window on Event Time
     *  处理延迟 Data 和 Watermarking
     *   *  Watermarking保障机制：
         *  Spark清除window聚合状态条件
         *   *  Update Mode
             *  Append Mode
     *  Join 操作
     *   *  Stream-static Joins
         *  Stream-stream Joins
         *   *  inner join
             *  outer join

# 简介 #

`Structured Streaming` 构建在`SparkSQL`之上的流处理引擎。可以使用户继续使用`DataSet/dataFrame`操  
作流数据。并且提供了多种计算模型可供选择，默认情况下，使用的依然是Spark的marco batch这种计  
算模型能够到100ms左右的end-to-end的精准一次的容错计算。除此之外也提供了基于`EventTime`语义  
的窗口计算（DStream 基于Processor Time不同）。同时在spark-2.3版本又提出新的计算模型  
`Continuous Processing`可以达到1ms左右的精准一次的容错计算。

# 快速入门案例 #

*  pom

<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-sql_2.11</artifactId>
    			<version>2.4.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-core_2.11</artifactId>
    			<version>2.4.3</version>
    		</dependency>

*  WordCount

def main(args: Array[String]): Unit = { 
    		Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    		Logger.getLogger("org.apache.jetty.server").setLevel(Level.OFF)
    
    		//1.构建SparkSession
    		val spark = SparkSession.builder()
    			.appName("wordcount")
    			.master("local[*]")
    			.getOrCreate()
    		import spark.implicits._
    
    		//2.创建输入流-readStream
    		var lines = spark.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 9999)
    			.load()
    
    		//3.对dataframe实现转换
    		var wordCounts = lines.as[String]
        		.flatMap(_.split("\\s+"))
        		.groupBy("value")
        		.count()
    
    
    		//4.构建query 输出
    		val query = wordCounts.writeStream
    			.format("console")
    			.outputMode(OutputMode.Update()) //有状态持续计算 Complete| Update| Append
    			.start()
    
    		//5.等待流结束
    		query.awaitTermination()
    	}

*  有状态持续计算 Complete| Update| Append 之间的区别

> 1.  Complete: 每一个trigger到来时，就输出整个完整的dataframe
> 2.  Update: 只输出那些被修改的Row。  
>     每一次window sliding，就去跟原来的结果比较，有变化就输出
> 3.  Append: 只输出新添加的（原来没有的）Row（）（如果是groupby，要有watermark才可以）  
>     每当一个watermark时间结束了，这个临时的结果再回转换成正式的结果并导出。

*  nc -l 999 输入

aa bb cc aa
    cc aa aa aa

*  输出结果(由于使用了Update 第二次输入没有`bb`，所有Batch: 2没有bb输出)

-------------------------------------------
    Batch: 1
    -------------------------------------------
    +-----+-----+
    |value|count|
    +-----+-----+
    |   cc|    1|
    |   bb|    1|
    |   aa|    2|
    +-----+-----+
    
    -------------------------------------------
    Batch: 2
    -------------------------------------------
    +-----+-----+
    |value|count|
    +-----+-----+
    |   cc|    2|
    |   aa|    5|
    +-----+-----+

## 程序流程结构 ##

> 1.构建`SparkSession` 对象  
> 2.借助于`SparkSession#readStream`加载动态的`Dataframe`  
> 3.使用`DataFrame API`或者是SQL语句 实现对动态数据计算  
> 4.通过`DataFrame#writeStream`方法构建`StreamQuery`对象  
> 5.调用`StreamQuery#awaitTermination`等待关闭指令

# 基本概念 #

Structure Stream的核心思想是通过将实时数据流看成是一个持续插入table.因此用户就可以使用SQL查 询DynamicTable|UnboundedTable。底层Spark通过StreamQuery实现对数据持续计算。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70]  
当对Input执行转换的时候系统产生一张结果表 `ResultTable` ,当有新的数据产生的时候，系统会往  
`Input Table` 插入一行数据，这会最终导致系统更新 `ResultTable` ,每一次的更新系统将更新的数  
据写到外围系统-Sink.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 1]

*  `Output` 定义如何将Result写出到外围系统，目前Spark支持三种输出模式：(上面已经简单介绍过了)

1.  `Complete Mode` \- 整个ResultTable的数据会被写到外围系统。
2.  `Update Mode` \- 只会将ResultTable中被更新的行，写到外围系统（ spark-2.1.1 +支持）
3.  `Append Mode` \- 只有新数据插入ResultTable的时候，才会将结果输出。注意：这种模式只适用  
    于被插入结果表的数据都是只读的情况下，才可以将输出模式定义为Append（查询当中不应该出  
    现聚合算子，当然也有特例，例如流中声明watermarker）

由于Structure Streaming计算的特点，Spark会在内存当中存储程序计算中间状态用于生产结果表的数  
据，Spark并不会存储 Input Table 的数据，一旦处理完成之后，读取的数据会被丢弃。整个聚合的  
过程无需用户干预（对比Storm，Storm状态管理需要将数据写到外围系统）。

# 故障容错 #

Structured Streaming通过`checkpoint`和`write ahead log`去记录每一次批处理的数据源的偏移量（区  
间），可以保证在失败的时候可以重复的读取数据源。其次Structure Streaming也提供了Sink的`幂等写`  
的特性（在编程中一个幂等 操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同），  
因此Structure Streaming实现`end-to-end` `exactly-once`语义的故障恢复。

# Structured Streaming API #

自Spark-2.0版本以后`Dataframe/Dataset`才可以处理有界数据和无界数据。`Structured Streaming`也是用  
`SparkSession`方式去创建`Dataset/DataFrame` ,同时所有`Dataset/DataFrame` 的操作保持和`Spark SQL`  
中`Dataset/DataFrame` 一致。

## Input Sources ##

### File Source ###

目前支持支持`text`, `csv`, `json`, `orc`, `parquet`等格式的文件，当这些数据被放入到采样目录，系统会以流的  
形式读取采样目录下的文件.

//1.创建SparkSession
    		val spark = SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		var df = spark.readStream
    			.format("text")
    			//json/csv/parquet/orc 等
    			.load("file:///Users/mashikang/IdeaProjects/structured_stream/src/main/resources")
    
    		var userDF = df.as[String]
    			.map(line => line.split("\\s+"))
    			.map(tokens => (tokens(0).toInt, tokens(1), tokens(2).toBoolean, tokens(3).toInt))
    			.toDF("id", "name", "sex", "age")
    
    		val query = userDF.writeStream.format("console")
    			.outputMode(OutputMode.Append())
    			.start()
    
    
    		query.awaitTermination()

*  文件

1 zhangsan true 20
    2 lisi true 28
    3 wangwu false 24
    4 zhaoliu true 28

#### Socket source(debug) ####

//1.构建SparkSession
    		val spark = SparkSession.builder()
    			.appName("wordcount")
    			.master("local[*]")
    			.getOrCreate()
    		import spark.implicits._
    
    		//2.创建输入流-readStream
    		var lines = spark.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 9999)
    			.load()
    
    		//3.对dataframe实现转换
    		var wordCounts = lines.as[String]
        		.flatMap(_.split("\\s+"))
        		.groupBy("value")
        		.count()
    
    
    		//4.构建query 输出
    		val query = wordCounts.writeStream
    			.format("console")
    			.outputMode(OutputMode.Update()) //有状态持续计算 Complete| Update
    			.start()
    
    		//5.等待流结束
    		query.awaitTermination()

### Kafka source ###

*  pom.xml

<dependency>
     	<groupId>org.apache.spark</groupId>
    	<artifactId>spark-sql-kafka-0-10_2.11</artifactId> 
    	<version>2.4.3</version> 
    </dependency>

//1.创建SparkSession
    		val spark=SparkSession .builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		var df=spark.readStream
    			.format("kafka")
    			.option("kafka.bootstrap.servers", "localhost:9092")
    			.option("subscribe", "topic01")
    			.load()
    			.selectExpr("CAST(key AS STRING)","CAST(value AS STRING)")
    
    		val wordCounts=df.select("value").as[String]
    			.flatMap(_.split("\\s+"))
    			.coalesce(1)
    			.map((_,1))
    			.toDF("word","count")
    			.groupBy("word")
    			.sum("count")
    
    		val query = wordCounts.writeStream.
    			format("console")
    			.outputMode(OutputMode.Update())
    			.start()
    
    		query.awaitTermination()

## Output Sink ##

### File sink(Append Mode Only) ###

//1.创建SparkSession
    		val spark = SparkSession.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		var df = spark.readStream
    			.format("kafka")
    			.option("kafka.bootstrap.servers", "localhost:9092")
    			.option("subscribe", "topic01")
    			.load()
    			.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    
    		val wordCounts = df.select("value").as[String]
    			.flatMap(_.split("\\s+"))
    			.coalesce(1)
    			.map((_, 1))
    			.toDF("word", "count")
    
    		val query = wordCounts.writeStream
    			.format("json")
    			.option("checkpointLocation", "file:///Users/mashikang/IdeaProjects/structured_stream/src/main/resources/checkpoints")
    			.outputMode(OutputMode.Append())
    			.start("file:Users/mashikang/IdeaProjects/structured_stream/src/main/resource/json")
    
    
    		query.awaitTermination()

### KafkaSink((Append|Update|Complete)) ###

//1.创建SparkSession
    		val spark = SparkSession.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		import org.apache.spark.sql.functions._
    
    		var df = spark.readStream
    			.format("kafka")
    			.option("kafka.bootstrap.servers", "localhost:9092")
    			.option("subscribe", "topic01")
    			.load()
    			.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    
    		val wordCounts = df.select("value").as[String]
    			.flatMap(_.split("\\s+"))
    			.coalesce(1)
    			.map((_, 1))
    			.toDF("word", "count")
    			.groupBy("word")
    			.agg(sum("count") as "count")
    			.selectExpr("word", "CAST(count AS STRING)")
    			.withColumnRenamed("word", "key")
    			.withColumnRenamed("count", "value")
    
    		val query = wordCounts.writeStream
    			.format("kafka")
    			.option("kafka.bootstrap.servers", "localhost:9092")
    			.option("topic", "topic02")
    			.option("checkpointLocation", "file:///Users/mashikang/IdeaProjects/structured_stream/src/main/resources/checkpoints")
    			.outputMode(OutputMode.Update())
    			.start()
    
    		query.awaitTermination()

### Foreach sink(Append|Update|Complate) ###

#### UserRowWriter ####

这里的 open方法在，每一次微批的时候触发，其中 `epochId`表示计算的批次。一般如果要保证  
`exactly-once` 语义的处理时候，需要在外围系统存储 `epochId`，如果存在重复计算 `epochId` 不  
变。

class UserRowWriter extends ForeachWriter[Row] { 
    	// 存储 上一次epochid
    	var lastEpochId: Long = -1L
    
    	/** * 计算 当前是否处理当前批次，如果epochId=lastEpochId说明是重复记录，丢弃更新 false * epochId!=lastEpochId 返回true 调用 open * * @param partitionId * @param epochId * @return */
    	override def open(partitionId: Long, epochId: Long): Boolean = { 
    		var flag: Boolean = false
    		if (epochId != -1L) { 
    			if (lastEpochId == epochId) { 
    				// 是重复记录
    				flag = false
    			} else { 
    				flag = true
    				lastEpochId = epochId
    			}
    		} else { 
    			// 第一次进来
    			lastEpochId = epochId
    			flag = true
    		}
    		flag
    	}
    
    	override def process(value: Row): Unit = { 
    		println(" ,epochId:" + lastEpochId)
    	}
    
    	override def close(errorOrNull: Throwable): Unit = { 
    		if (errorOrNull != null)
    			errorOrNull.printStackTrace()
    	}
    }

//1.创建SparkSession
    		val spark = SparkSession.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		import org.apache.spark.sql.functions._
    
    		var df = spark.readStream
    			.format("kafka")
    			.option("kafka.bootstrap.servers", "localhost:9092")
    			.option("subscribe", "topic01")
    			.load()
    			.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    
    		val wordCounts = df.select("value").as[String]
    			.flatMap(_.split("\\s+"))
    			.coalesce(1)
    			.map((_, 1))
    			.toDF("word", "count")
    			.groupBy("word")
    			.agg(sum("count") as "count")
    			.selectExpr("word", "CAST(count AS STRING)")
    			.withColumnRenamed("word", "key")
    			.withColumnRenamed("count", "value")
    
    		val query = wordCounts.writeStream
    			.outputMode(OutputMode.Update())
    			.foreach(new UserRowWriter)
    			.start()
    
    		query.awaitTermination()

## Window on Event Time ##

Structured Streaming使用聚合函数基于EventTime计算window是非常简单的类似于分组聚合。分组聚  
合是按照指定的column字段对表中的数据进行分组，然后使用聚合函数对用户指定的column字段进行  
聚合。  
下面一张图描绘的是计算10分钟内的单词统计，每间隔5分钟滑动一个时间窗口。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 2]  
按照窗口原始含义是将落入到同一个窗口的数据进行分组，因此在Structured Streaming可以使用  
groupby和window表达窗口计算

Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    		Logger.getLogger("org.apache.jetty.server").setLevel(Level.OFF)
    
    		//1.创建SparkSession
    		val spark=SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		//字符,时间戳
    		var df=spark.readStream .format("socket")
    			.option("host", "localhost")
    			.option("port", "9999")
    			.load()
    
    		import org.apache.spark.sql.functions._
    		var sdf=new SimpleDateFormat("mm:ss")
    
    		val wordCounts=df.select("value")
    			.as[String]
    			.map(_.split(","))
    			// 这里的Timestamp导如java.sql的依赖
    			.map(tokens=>(tokens(0),new Timestamp(tokens(1).toLong)))
    			.toDF("word","timestamp")
    			.groupBy(
    				window($"timestamp","10 seconds","5 seconds"),
    				$"word"
    			)
    			.count()
    			.map(r=>
    				(sdf.format(r.getStruct(0).getTimestamp(0)),
    					sdf.format(r.getStruct(0).getTimestamp(1)),
    					r.getString(1),r.getLong(2)))
    			.toDF("start","end","word","count")
    
    		val query = wordCounts.writeStream
    			.outputMode(OutputMode.Update())
    			.format("console")
    			.start()
    
    
    		query.awaitTermination()

## 处理延迟 Data 和 Watermarking ##

默认情况下，Spark会把落入到时间窗口的数据进行聚合操作。但是需要思考的是Event-Time是基于事  
件的时间戳进行窗口聚合的。那就有可能事件窗口已经触发很久了，但是有一些元素因为某种原因，导  
致迟到了，这个时候Spark需要将迟到的的数据加入到已经触发的窗口进行重复计算。但是需要注意如  
果在长时间的流计算过程中，如果不去限定窗口计算的时间，那么意味着Spark要在内存中一直存储窗  
口的状态，这样是不切实际的，因此Spark提供一种称为`watermarker`的机制用于限定存储在Spark内存  
中中间结果存储的时间，这样系统就可以将已经确定触发过的窗口的中间结果给删除。如果后续还有数  
据在窗口endtime以后抵达该窗口，Spark把这种数据定义为`late数据`。其中`watermarker`计算方式 `max event time seen by engine - late threshold`如果`watermarker`的取值大于了时间窗口的  
endtime即可认定该窗口的计算结果就可以被丢弃了。如果此时再有数据落入到已经被丢弃的时间窗  
口，则该迟到的数据会被Spark放弃更新，也就是丢弃。

> Watermarking=`max event time seen by engine - late threshold`

### Watermarking保障机制： ###

*  能够保证在window的EndTime > 水位线的窗口的状态Spark会存储起来，这个时候如果有迟到的  
    数据再水位线没有淹没window之前Spark可以保障迟到的数据能正常的处理。
 *  如果水位线已经没过了窗口的end时间，那么后续迟到数据不一定能够被处理，换句话说，迟到越  
    久的数据 被处理的几率越小。

> 如果是使用`水位线计算` ，输出模式必须是Update或者Append,否则系统不会删除。

Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    		Logger.getLogger("org.apache.jetty.server").setLevel(Level.OFF)
    
    		//1.创建SparkSession
    		val spark=SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    		//字符,时间戳
    		var df=spark.readStream .format("socket")
    			.option("host", "localhost")
    			.option("port", "9999")
    			.load()
    
    		import org.apache.spark.sql.functions._
    		var sdf=new SimpleDateFormat("mm:ss")
    
    		val wordCounts=df.select("value")
    			.as[String]
    			.map(_.split(","))
    			// 这里的Timestamp导如java.sql的依赖
    			.map(tokens=>(tokens(0),new Timestamp(tokens(1).toLong)))
    			.toDF("word","timestamp")
    			// 与上面窗口的API相比，多了水位线的设置
    			.withWatermark("timestamp", "5 seconds")
    			.groupBy(
    				window($"timestamp","10 seconds","5 seconds"),
    				$"word"
    			)
    			.count()
    			.map(r=>
    				(sdf.format(r.getStruct(0).getTimestamp(0)),
    					sdf.format(r.getStruct(0).getTimestamp(1)),
    					r.getString(1),r.getLong(2)))
    			.toDF("start","end","word","count")
    
    		val query = wordCounts.writeStream
    			.outputMode(OutputMode.Update())
    			.format("console")
    			.start()
    
    
    		query.awaitTermination()

### Spark清除window聚合状态条件 ###

*  Output mode 必须是 Append 或者 Update.，如果是Update 只要窗口有数据更新即可有输出。  
    如果是Append，必须当水位线没过window的时候才会将Result写出。

#### Update Mode ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 3]

#### Append Mode ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 4]

*  必须在分组出现聚合使用时间column/window列
 *  withWaterMaker的时间column必须和groupBy后面时间column保持一致，例如： 错误实例`df.withWatermark("time", "1 min").groupBy("time2").count()`。
 *  一定要在分组聚合之前调用withWaterMaking，例如`df.groupBy("time").count().withWatermark("time", "1 min")` 错误实例  
    `df.withWatermark("time", "1 min").groupBy("time").count()`正确写法。

## Join 操作 ##

Structured Streaming 不仅仅支持对静态的 Dataset/DataFrame 做join操作，也支持对streaming  
Dataset/DataFrame实现join操作。

*  Stream-static Joins `spark-2.0` 支持
 *  Stream-stream Joins `Spark 2.3` 支持

### Stream-static Joins ###

//1.创建SparkSession
    		val spark=SparkSession
    			.builder()
    			.master("local[6]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    
    		import org.apache.spark.sql.functions._
    
    		/** * +---+------+---+ * | id| name|age| * +---+------+---+ * | 1| 张三 | 18| * | 2| lisi| 28| * | 3|wangwu| 38| * +---+------+---+ */
    		val userDF=spark.read
    			.format("json")
    			.load("/Users/mashikang/IdeaProjects/structured_stream/src/main/resources/json")
    			.selectExpr("CAST(id AS INTEGER)","name","CAST(age AS INTEGER)")
    
    		//1 apple 1 4.5
    		var orderItemDF= spark.readStream
    			.format("socket")
    			.option("host","localhost")
    			.option("port",9999)
    			.load() .as[String]
    			.map(line=>line.split("\\s+"))
    			.map(tokens=>(tokens(0).toInt,tokens(1),tokens(2).toInt,tokens(3).toDouble))
    			.toDF("uid","item","count","price")
    
    		val jointResults = orderItemDF.join(userDF,$"id"===$"uid","left_outer")
    
    		val query = jointResults
    			.writeStream
    			.format("console")
    			.outputMode(OutputMode.Append())
    			.start()
    
    		query.awaitTermination()

### Stream-stream Joins ###

>  *  两边流都必须声明watermarker，告知引擎什么是可以清楚状态（默认取最低）。
>  *  需要在连接条件中添加eventTime column的时间约束，这样引擎就知道什么时候可以清除后续  
>     的流的状态。
>     
>      *  Time range join conditions
>      *  Join on event-time windows

#### inner join ####

*  **方案1 Time range join conditions**

//1.创建SparkSession
    		val spark = SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    
    		//001 apple 1 4.5 1566529410000
    		val orderDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 9999)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), (t(2).toInt * t(3).toDouble), new Timestamp(t(4).toLong)))
    			.toDF("uid", "item", "cost", "order_time")
    
    		//001 zhangsan 1566529410000
    		val userDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 8888)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), new Timestamp(t(2).toLong)))
    			.toDF("id", "name", "login_time")
    
    		import org.apache.spark.sql.functions._
    
    		//用户的登陆数据缓存 2 seconds 订单数据缓存4秒
    		val userWatermarker = userDF.withWatermark("login_time", "2 seconds")
    		val orderWaterMarker = orderDF.withWatermark("order_time", "4 seconds")
    
    		//连接 用户登陆以后将2秒以内的购买行为和用进行join 
    		val joinDF = userWatermarker.join(orderWaterMarker,
    			expr(
    				"""
    				  |id=uid and order_time >= login_time and order_time <= login_time + interval 2 seconds 
    				""".stripMargin)
    		)
    		val query = joinDF.writeStream
    			.format("console")
    			.outputMode(OutputMode.Append()).start() 
    		
    		query.awaitTermination()

*  **方案2Join on event-time windows**

//1.创建SparkSession
    		val spark = SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    
    		//001 apple 1 4.5 1566529410000
    		val orderDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 9999)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), (t(2).toInt * t(3).toDouble), new Timestamp(t(4).toLong)))
    			.toDF("uid", "item", "cost", "order_time")
    
    		//001 zhangsan 1566529410000
    		val userDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 8888)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), new Timestamp(t(2).toLong)))
    			.toDF("id", "name", "login_time")
    
    		import org.apache.spark.sql.functions._
    
    		//用户的登陆数据缓存 2 seconds 订单数据缓存4秒
    		val userWatermarker = userDF.withWatermark("login_time", "2 seconds")
    			.select(
    				window($"login_time", "5 seconds"),
    				$"id", $"name", $"login_time")
    			.withColumnRenamed("window", "leftWindow")
    
    		val orderWaterMarker = orderDF.withWatermark("order_time", "4 seconds")
    			.select(
    				window($"order_time", "5 seconds"),
    				$"uid", $"item", $"cost", $"order_time")
    			.withColumnRenamed("window", "rightWindow")
    
    		//连接用户登陆以后将2秒以内的购买行为和用进行join
    		val joinDF = userWatermarker
    			.join(
    				orderWaterMarker,
    				expr(
    					"""
    					  |id=uid and leftWindow = rightWindow
    					""".stripMargin)
    			)
    			
    		val query = joinDF.writeStream
    			.format("console")
    			.outputMode(OutputMode.Append()).start() 
    		
    		query.awaitTermination()

#### outer join ####

//1.创建SparkSession
    		val spark = SparkSession
    			.builder()
    			.master("local[*]")
    			.appName("printline")
    			.getOrCreate()
    
    		import spark.implicits._
    
    		//001 apple 1 4.5 1566529410000
    		val orderDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 9999)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), (t(2).toInt * t(3).toDouble), new Timestamp(t(4).toLong)))
    			.toDF("uid", "item", "cost", "order_time")
    
    		//001 zhangsan 1566529410000
    		val userDF = spark
    			.readStream
    			.format("socket")
    			.option("host", "localhost")
    			.option("port", 8888)
    			.load()
    			.map(row => row.getAs[String]("value").split("\\s+"))
    			.map(t => (t(0), t(1), new Timestamp(t(2).toLong)))
    			.toDF("id", "name", "login_time")
    
    		import org.apache.spark.sql.functions._
    
    		//系统分别会对 user 和 order 缓存 最近 1 seconds 和 2 seconds 数据,
    		// 一旦时间过去，系统就无 法保证数据状态继续保留
    		val loginWatermarker=userDF.withWatermark("login_time","1 second")
    		val orderWatermarker=orderDF.withWatermark("order_time","2 seconds")
    
    		//计算订单的时间 & 用户 登陆之后的0~1 seconds 关联 数据 并且进行join
    		val joinDF = loginWatermarker
    		.join(
    				orderWatermarker,
    				expr("uid=id and order_time >= login_time and order_time <= login_time + interval 1 seconds"),
    				"leftOuter"
    			)
    			
    		val query = joinDF.writeStream
    			.format("console")
    			.outputMode(OutputMode.Append()).start() 
    		
    		query.awaitTermination()

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200612174115861.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200612174328732.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200615160910610.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200615175338128.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200615175542447.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg=,size_16,color_FFFFFF,t_70