2.3 编程模型

Structured Streaming将流式数据当成一个不断增长的table，然后使用和批处理同一套API，都是基于DataSet/DataFrame的。如下图所示，通过将流式数据理解成一张不断增长的表，从而就可以像操作批的静态数据一样来操作流数据了。

在这个模型中，主要存在下面几个组成部分：

第一部分：Input Table（Unbounded Table），流式数据的抽象表示，没有限制边界的，表的

数据源源不断增加；

第二部分：Query（查询），对 Input Table 的增量式查询，只要Input Table中有数据，立即（默认情况）执行查询分析操作，然后进行输出（类似SparkStreaming中微批处理）；

第三部分：Result Table，Query 产生的结果表；

第四部分：Output，Result Table 的输出，依据设置的输出模式OutputMode输出结果；

Structured Streaming最核心的思想就是将实时到达的数据看作是一个不断追加的unboundtable无界表，到达流的每个数据项就像是表中的一个新行被附加到无边界的表中，用静态结构化数据的批处理查询方式进行流计算。

以词频统计WordCount案例，Structured Streaming实时处理数据的示意图如下，各行含义：

第一行、表示从TCP Socket不断接收数据，使用【nc -lk 9999】；

第二行、表示时间轴，每隔1秒进行一次数据处理；

第三行、可以看成是“input unbound table"，当有新数据到达时追加到表中；

第四行、最终的wordCounts是结果表，新数据到达后触发查询Query，输出的结果；

第五行、当有新的数据到达时，Spark会执行“增量"查询，并更新结果集；该示例设置为CompleteMode，因此每次都将所有数据输出到控制台；

上图中数据实时处理说明：

第一、在第1秒时，此时到达的数据为"cat dog"和"dog dog"，因此可以得到第1秒时的结果集cat=1dog=3，并输出到控制台；

第二、当第2秒时，到达的数据为"owl cat"，此时"unbound table"增加了一行数据"owl cat"，执

行word count查询并更新结果集，可得第2秒时的结果集为cat=2 dog=3 owl=1，并输出到控制台；

第三、当第3秒时，到达的数据为"dog"和"owl"，此时"unbound table"增加两行数据"dog"和

“owl”，执行word count查询并更新结果集，可得第3秒时的结果集为cat=2 dog=4 owl=2；使用Structured Streaming处理实时数据时，会负责将新到达的数据与历史数据进行整合，并完成正确的计算操作，同时更新Result Table。

3 入门案例：WordCount

入门案例与SparkStreaming的入门案例基本一致：实时从TCP Socket读取数据（采用nc）实时进行词频统计WordCount，并将结果输出到控制台Console。

文档：http://spark.apache.org/docs/2.4.5/structured-streaming-programming-guide.html#quick-example

3.1 功能演示

运行词频统计WordCount程序，从TCP Socket消费数据，官方演示说明截图如下：

演示运行案例步骤：

第一步、打开终端Terminal，运行NetCat，命令为：nc -lk 9999

第二步、打开另一个终端Terminal，执行如下命令

# 官方入门案例运行：词频统计
/export/server/spark/bin/run-example \
--master local[2] \
--conf spark.sql.shuffle.partitions=2 \
org.apache.spark.examples.sql.streaming.StructuredNetworkWordCount \
node1.oldlu.cn 9999
# 测试数据
spark hadoop spark hadoop spark hive
spark spark spark
spark hadoop hive

发送数据以后，最终统计输出结果如下：

3.2 Socket 数据源

从Socket中读取UTF8文本数据。一般用于测试，使用nc -lk 端口号向Socket监听的端口发送数据，用于测试使用，有两个参数：

参数一：host，主机名称，必须指定参数

参数二：port，端口号，必须指定参数

范例如下所示：

3.3 Console 接收器

将结果数据打印到控制台或者标准输出，通常用于测试或Bedug使用，三种输出模式OutputMode（Append、Update、Complete）都支持，两个参数可设置：

参数一：numRows，打印多少条数据，默认为20条；

参数二：truncate，如果某列值字符串太长是否截取，默认为true，截取字符串；

范例如下所示：

3.4 编程实现

可以认为Structured Streaming = SparkStreaming + SparkSQL，对流式数据处理使用SparkSQL数据结构，应用入口为SparkSession，对比SparkSQL与SparkStreaming编程：

Spark Streaming：将流式数据按照时间间隔（BatchInterval）划分为很多Batch，每批次数据封装在RDD中，底层RDD数据，构建StreamingContext实时消费数据；

Structured Streaming属于SparkSQL模块中一部分，对流式数据处理，构建SparkSession对象，

指定读取Stream数据和保存Streamn数据，具体语法格式：

加载数据Load：读取静态数据【spark.read】、读取流式数据【spark.readStream】

保存数据Save：保存静态数据【ds/df.write】、保存流式数据【ds/df.writeStrem】词频统计案例：从TCP Socket实消费流式数据，进行词频统计，将结果打印在控制台Console 。

第一点、程序入口SparkSession，加载流式数据：spark.readStream
第二点、数据封装Dataset/DataFrame中，分析数据时，建议使用DSL编程，调用API，很少使用SQL方式
第三点、启动流式应用，设置Output结果相关信息、start方法启动应用

完整案例代码如下：

import org.apache.spark.sql.streaming.{OutputMode,StreamingQuery}
import org.apache.spark.sql.{DataFrame,SparkSession}
/**
 * 使用Structured Streaming从TCP Socket实时读取数据，进行词频统计，将结果打印到控制台。
 */
        object StructuredWordCount{
        def main(args:Array[String]):Unit={
// TODO: 构建SparkSession实例对象
        val spark:SparkSession=SparkSession.builder()
        .appName(this.getClass.getSimpleName.stripSuffix("$"))
        .master("local[2]")
        .config("spark.sql.shuffle.partitions","2") // 设置Shuffle分区数目
        .getOrCreate()
// 导入隐式转换和函数库
import spark.implicits._
import org.apache.spark.sql.functions._
// TODO: 1. 从TCP Socket 读取数据
val inputStreamDF:DataFrame=spark.readStream
        .format("socket")
        .option("host","node1.oldlu.cn").option("port",9999)
        .load()
/*
root
|-- value: string (nullable = true)
*/
//inputStreamDF.printSchema()
// TODO: 2. 业务分析：词频统计WordCount
        val resultStreamDF:DataFrame=inputStreamDF
        .as[String] // 将DataFrame转换为Dataset进行操作
// 过滤数据
        .filter(line=>null!=line&&line.trim.length>0)
// 分割单词
        .flatMap(line=>line.trim.split("\\s+"))
        .groupBy($"value").count() // 按照单词分组，聚合
/*
root
|-- value: string (nullable = true)
|-- count: long (nullable = false)
*/
//resultStreamDF.printSchema()
// TODO: 3. 设置Streaming应用输出及启动
        val query:StreamingQuery=resultStreamDF.writeStream
// TODO: 设置输出模式：Complete表示将ResultTable中所有结果数据输出
// .outputMode(OutputMode.Complete())
// TODO: 设置输出模式：Update表示将ResultTable中有更新结果数据输出
        .outputMode(OutputMode.Update())
        .format("console")
        .option("numRows","10").option("truncate","false")
// 流式应用，需要启动start
        .start()
// 流式查询等待流式应用终止
        query.awaitTermination()
// 等待所有任务运行完成才停止运行
        query.stop()
        }
     }

其中可以设置不同输出模式（OutputMode），当设置为Complete时，结果表ResultTable中

所有数据都输出；当设置为Update时，仅仅输出结果表ResultTable中更新的数据。

4 DataStreamReader 接口

从Spark 2.0至Spark 2.4版本，目前支持数据源有4种，其中Kafka 数据源使用作为广泛，其他数据源主要用于开发测试程序。

文档： http://spark.apache.org/docs/2.4.5/structured-streaming-programming-guide.html#input-sources

在Structured Streaming中使用SparkSession#readStream读取流式数据，返回DataStreamReader对象，指定读取数据源相关信息，声明如下：

查看DataStreamReader中方法可以发现与DataFrameReader中基本一致，编码上更加方便加载流式数据。

5 文件数据源

将目录中写入的文件作为数据流读取，支持的文件格式为：text、csv、json、orc、parquet，可以设置相关可选参数：

从文件数据源加载数据伪代码如下：

val streamDF = spark
        .readStream
// Schema must be specified when creating a streaming source DataFrame.
        .schema(schema)
// 每个trigger最大文件数量
        .option("maxFilesPerTrigger",100)
// 是否首先计算最新的文件，默认为false
        .option("latestFirst",value = true)
// 是否只检查名字，如果名字相同，则不视为更新，默认为false
        .option("fileNameOnly",value = true)
        .csv("*.csv")

演示范例：监听某一个目录，读取csv格式数据，统计年龄小于25岁的人群的爱好排行榜。

测试数据

jack;23;running
charles;32;basketball
tom;28;football
lili;24;running
bob;20;swimming
zhangsan;32;running
lisi;28;running
wangwu;24;running
zhaoliu;26;swimming
honghong;28;running`

业务实现代码，监控Windows系统目录【D:/datas】

import org.apache.spark.sql.streaming.{OutputMode,StreamingQuery}
import org.apache.spark.sql.types.{IntegerType,StringType,StructType}
import org.apache.spark.sql.{DataFrame,Dataset,Row,SparkSession}
/**
 * 使用Structured Streaming从目录中读取文件数据：统计年龄小于25岁的人群的爱好排行榜
 */
        object StructuredFileSource{
        def main(args:Array[String]):Unit={
// 构建SparkSession实例对象
        val spark:SparkSession=SparkSession.builder()
        .appName(this.getClass.getSimpleName.stripSuffix("$"))
        .master("local[2]")
// 设置Shuffle分区数目
        .config("spark.sql.shuffle.partitions","2")
        .getOrCreate()
// 导入隐式转换和函数库
import spark.implicits._
import org.apache.spark.sql.functions._
// TODO: 从文件系统，监控目录，读取CSV格式数据
// 数据样本 -> jack,23,running
val csvSchema:StructType=new StructType()
        .add("name",StringType,nullable=true)
        .add("age",IntegerType,nullable=true)
        .add("hobby",StringType,nullable=true)
        val inputStreamDF:DataFrame=spark.readStream
        .option("sep",";")
        .option("header","false")
// 指定schema信息
        .schema(csvSchema)
        .csv("file:///D:/datas/")
// 依据业务需求，分析数据：统计年龄小于25岁的人群的爱好排行榜
        val resultStreamDF:Dataset[Row]=inputStreamDF
// 年龄小于25岁
        .filter($"age"< 25)
// 按照爱好分组统计
        .groupBy($"hobby").count()
// 按照词频降序排序
        .orderBy($"count".desc)
// 设置Streaming应用输出及启动
        val query:StreamingQuery=resultStreamDF.writeStream
// 对流式应用输出来说，设置输出模式
        .outputMode(OutputMode.Complete())
        .format("console")
        .option("numRows","10")
        .option("truncate","false")
// 流式应用，需要启动start
        .start()
// 查询器等待流式应用终止
        query.awaitTermination()
        query.stop() // 等待所有任务运行完成才停止运行
        }
   }

6 Rate source

以每秒指定的行数生成数据，每个输出行包含2个字段：timestamp和value。其中timestamp是一个Timestamp含有信息分配的时间类型，并且value是Long（包含消息的计数从0开始作为第一行）类型。此源用于测试和基准测试，可选参数如下：

演示范例代码如下：

import org.apache.spark.sql.streaming.{OutputMode,StreamingQuery,Trigger}
import org.apache.spark.sql.{DataFrame,SparkSession}
/**
 * 数据源：Rate Source，以每秒指定的行数生成数据，每个输出行包含一个timestamp和value。
 */
        object StructuredRateSource{
        def main(args:Array[String]):Unit={
// 构建SparkSession实例对象
        val spark:SparkSession=SparkSession.builder()
        .appName(this.getClass.getSimpleName.stripSuffix("$"))
        .master("local[2]")
// 设置Shuffle分区数目
        .config("spark.sql.shuffle.partitions","2")
        .getOrCreate()
// 导入隐式转换和函数库
import spark.implicits._
import org.apache.spark.sql.functions._
// TODO：从Rate数据源实时消费数据
val rateStreamDF:DataFrame=spark.readStream
        .format("rate")
        .option("rowsPerSecond","10") // 每秒生成数据数目
        .option("rampUpTime","0s") // 每条数据生成间隔时间
        .option("numPartitions","2") // 分区数目
        .load()
/*
root
|-- timestamp: timestamp (nullable = true)
|-- value: long (nullable = true)
*/
//rateStreamDF.printSchema()
// 3. 设置Streaming应用输出及启动
        val query:StreamingQuery=rateStreamDF.writeStream
// 设置输出模式：Append表示新数据以追加方式输出
        .outputMode(OutputMode.Append())
        .format("console")
        .option("numRows","10")
        .option("truncate","false")
// 流式应用，需要启动start
        .start()
// 流式查询等待流式应用终止
        query.awaitTermination()
// 等待所有任务运行完成才停止运行
        query.stop()
        }
    }

运行应用程序，随机生成的数据，截图如下：

大数据Spark Structured Streaming 2

2.3 编程模型

3 入门案例：WordCount

3.1 功能演示

3.2 Socket 数据源

3.3 Console 接收器

3.4 编程实现

4 DataStreamReader 接口

5 文件数据源

6 Rate source

热门文章

最新文章

相关课程

相关电子书

相关实验场景