一、实验目的
掌握流式DataFrame的选择、投影和聚合操作。
掌握流式DataFrame的JOIN连接操作。
二、实验内容
1、读取json格式的文件数据源，并执行转换操作。
2、连接两个DataFrame。
三、实验原理
使用流数据格式的DataFrame，可以将任何select和filter转换应用到它，以及应用任何在个别列上操作的Spark SQL函数。此外，基本聚合和高级分析函数也可用于流DataFrame。
在流DataFrame中，不支持以下DataFrame转换，因为它们太过复杂，无法维护状态，或者由于流数据的无界性。
• 在流DataFrame上的多个聚合或聚合链。
• limit和take N行。
• distinct转换。但是，有一种方法可以使用唯一标识符来删除重复数据。
• 在没有任何聚合的情况下对流DataFrame进行排序。然而，在某种形式的聚合之后排序是得到支持的。
四、实验环境
硬件：x86_64 ubuntu 16.04服务器
软件：JDK 1.8.162,Spark-2.3.2,Hadoop-2.7.3
五、实验步骤
5.1 启动Spark集群
1、在终端窗口下，输入以下命令，分别启动Spark集群：
1. $ start-dfs.sh 2. $ cd /opt/spark 3. $ ./sbin/start-all.sh
2、在HDFS上面创建文件
1. hdfs dfs -mkdir -p /data/dataset/streaming
3、启动spark-shell。在终端窗口下，执行如下命令：（注意，请将以下命令中的localhost替换为虚拟机实际的机器名）
1. $ spark-shell --master spark://localhost:7077
5.2 数据源说明
创建一个数据源文件，文件存放位置位于”/data/dataset/streaming/“目录下。在终端窗口下，执行以下命令:
1. $ cd /data/dataset/streaming/ 2. $ vi language.json 文件内容如下： 1. {"name":"python","age":12} 2. {"name":"scala","age":23} 3. {"name":"python","age":15}
5.3 结构化流DataFrame基本操作：投影和过滤
1、首先创建一个Schema。在spark-shell窗口下，分别执行如下代码：
1. import org.apache.spark.sql.types._ 2. 3. val personSchema = new StructType().add("name", StringType,false).add("age", IntegerType,false)
2、读取文件数据源。在spark-shell窗口下，分别执行如下代码：
1. val data = spark.readStream.schema(personSchema).json("hdfs://localhost:9000/data/dataset/streaming/")
3、过滤和投影。选择查看数据中的name 和 age 字段，并且过滤 age 大于 12的人。在spark-shell窗口下，执行如下代码：
1. val result1 = data.select($"name",$"age").where($"age" > 12)
4、将流查询结果输出到控制台，并启动流程序。在spark-shell窗口下，执行如下代码：
1. val query1 = result1.writeStream.format("console").outputMode("append").start()
5、将本地的/data/dataset/language.json文件上传到HDFS受流程序监视的目录中。另打开一个终端，在该终端窗口下，执行如下命令：
1. $ hdfs dfs -put /data/dataset/streaming/language.json /data/dataset/streaming/
6. 切换回spark-shell窗口，可以看到如下的输出结果:
7、结束流程序运行。在spark-shell窗口下，执行如下代码：
1. query1.stop
5.4 结构化流DataFrame基本操作：数据聚合操作
1、删除HDFS目录下的文件
1. $ hdfs dfs -rm /data/dataset/streaming/language.json
2、按name字段对数据进行分组，并分组对age字段进行求和。在spark-shell窗口下，执行如下代码：
1. import org.apache.spark.sql.types._ 2. 3. val personSchema = new StructType()add("name", StringType,false).add("age", IntegerType,false) 4. 5. val data = spark.readStream.schema(personSchema).json("hdfs://localhost:9000/data/dataset/streaming") 6. 7. val result2 = words.groupBy($"name").sum("age") 8. 9. val query2 = result2.writeStream.format("console").outputMode("complete").start()
3、将本地的/data/dataset/language.json文件上传到HDFS受流程序监视的目录中。另打开一个终端，在该终端窗口下，执行如下命令：
1. $ hdfs dfs -put /data/dataset/streaming/language.json /data/dataset/streaming/
4. 切换回spark-shell窗口，可以看到如下的输出结果:
由上面结果可以看出对两条python的数据中的age值进行了求和。
5、结束流程序运行。在spark-shell窗口下，执行如下代码：
1. query2.stop
5.3 连接两个流DataFrame
可以用一个streaming DataFrame来做最酷的事情之一，就是join一个静态的DataFrame或者另一个streaming DataFrame。从Spark 2.3开始，结构化流支持join两个streaming DataFrames。
1、删除HDFS目录下的文件
1. $ hdfs dfs -rm /data/dataset/streaming/language.json
2、读取流数据。在spark-shell窗口下，执行如下代码：
1. import org.apache.spark.sql.types._ 2. 3. val personSchema = new StructType().add("name", StringType,false).add("age", IntegerType,false) 4. 5. val dataDF = spark.readStream.schema(personSchema).json("hdfs://localhost:9000/data/dataset/streaming")
3、通过join把两个DataFrame连接在一起（按name字段进行连接）。在spark-shell窗口下，执行如下代码：
1. val df = dataDF.join(dataDF, "name")
4、输出连接后的DataFrame。在spark-shell窗口下，执行如下代码：
1. val query = df.writeStream.format("console").outputMode("append").start()
5、将本地的/data/dataset/language.json文件上传到HDFS受流程序监视的目录中。另打开一个终端，在该终端窗口下，执行如下命令：
1. $ hdfs dfs -put /data/dataset/streaming/language.json /data/dataset/streaming/
6. 切换回spark-shell窗口，可以看到如下的输出结果:
7、结束流程序运行。在spark-shell窗口下，执行如下代码：
1. query.stop
— END —
