Spark流计算期末复习资料----助力高绩点

ゝ一纸荒年。 2023-07-15 09:17 6阅读 0赞

**Spark****集群搭建**

Spark基于standalone集群搭建，standalone是主从结构，分master,worker;app作业

Application：带有自己需要的mem和cpu资源量，会在master里排队，最后被分发到worker上执行。app的启动是去各个worker遍历，获取可用的cpu，然后去各个worker launch executor。

Worker：每台slave起一个(也可以起多个)，默认或被设置cpu和mem数，并在内存里做加减维护资源剩余量。Worker同时负责拉起本地的executor backend，即执行进程。

Master：接受Worker、app的注册，为app执行资源分配。Master和Worker本质上都是一个带Actor的进程

1.1 解压安装包。在命令终端切换到/simple目录下，分别执行如下命令：

cd /simple

tar –zxvf /simple/soft /scala-2.11.4.tgz

tar –zxvf /simple/soft/spark-1.6.1-bin-hadoop2.4.tgz

1.2 环境变量配置。在任意目录下执行命令：vim /etc/profile编辑配置文件。

JAVA\_HOME=/simple/jdk1.7.0\_79

HADOOP\_HOME=/simple/hadoop-2.4.1

export SCALA\_HOME=/simple/scala-2.11.4

export SPARK\_HOME=/simple/spark-1.6.1-bin-hadoop2.4

export PATH=$JAVA\_HOME/bin:$HADOOP\_HOME/sbin:$SCALA\_HOME/bin:$SPARK\_HOME/sbin:$SPARK\_HOME/bin:$PATH

1.3 环境变量配置文件生效，执行命令：source /etc/profile

1.4 查看本机IP，在任意目录下执行命令：ifconfig查看网卡信息。

2.1 切换至spark安装目录下的/simple/spark-1.6.1-bin-hadoop2.4/conf文件并执行命令：

cp spark-env.sh.template spark-env.sh

cp slaves.template slaves

2.2 配置spark-env.sh。

export SPART\_MASTER\_IP=192.168.1.2

export SPART\_MASTER\_PORT=7077

export SPART\_WORKER\_CORES=2

export SPART\_WORKER\_INSTANCES=1

export SPART\_WORKER\_MEMORY=1024M

export JAVA\_HOME=/simple/jdk1.7.0\_79

export SCALA\_HOME=/simple/scala-2.11.4

2.3 配置slaves: 输入spark02(主机名)

2.4 启动集群，首先切换到目录：cd /simple/spark-1.6.1-bin-hadoop2.4/sbin

执行启动脚本：./start-all.sh并查看spark进程。

2.5 集群启动是否成功，启动浏览器并输入地址 simple:8080

**统计文本每个单词出现的次数并对统计结果进行排序操作**

Spark，是一种"One Stack to rule them all"的大数据计算框架，期望使用一个技术堆栈就完美地解决大数据领域的各种计算任务。Apache官方，对Spark的定义就是：通用的大数据快速处理引擎。（基于内存）

Spark使用Spark RDD、Spark SQL、Spark Streaming、MLlib、GraphX成功解决了大数据领域中，离线批处理、交互式查询、实时流计算、机器学习与图计算等最重要的任务和问题。

Spark除了一站式的特点之外，另外一个最重要的特点，就是基于内存进行计算，从而让它的速度可以达到MapReduce、Hive的数倍甚至数十倍！现在已经有很多大公司正在生产环境下深度地使用Spark作为大数据的计算框架，包括eBay、Yahoo!、BAT、网易、京东、华为、大众点评、优酷土豆、搜狗等等。

2.1在/simple/目录下使用命令:touch spark.txt创建名为spark.txt的文本数据，数据信息如下：

hello world me you world

me you say hello world

Hello me wei you wei me

2.2编辑上面创建的对象“SortWordCout”，然后编辑内容如下：

**object** SortWordCout \{

**def** main(args: Array\[*String*\]) \{

// 创建SparkConf和JavaSparkContext

**val** conf = **new** SparkConf().setAppName("SortWordCount").setMaster("local")

**val** sc = **new** SparkContext(conf)

// 创建lines RDD

**val** lines = sc.textFile("/simple/spark.txt", 1)

// 执行我们之前做过的单词计数

**val** words = lines.flatMap \{ line => line.split(" ") \}

**val** pairs = words.map \{ word => (word,1) \}

**val** wordCounts = pairs.reduceByKey(\_+\_)

// 到这里为止，就得到了每个单词出现的次数

// 但是，问题是，我们的新需求，是要按照每个单词出现次数的顺序，降序排序

// wordCounts RDD内的元素是什么？应该是这种格式的吧：(hello, 3) (you, 2)

// 我们需要将RDD转换成(3, hello) (2, you)的这种格式，才能根据单词出现次数进行排序把！

// 进行key-value的反转映射

**val** countWords = wordCounts.map(wordCount => (wordCount.\_2,wordCount.\_1))

// 按照key进行排序

**val** sortedCountWords = countWords.sortByKey(**false**)

// 再次将value-key进行反转映射

**val** sortedWordCounts = sortedCountWords.map(sortedCountWord =>(sortedCountWord.\_2,sortedCountWord.\_1))

// 到此为止，我们获得了按照单词出现次数排序后的单词计数，打印出来

sortedWordCounts.foreach(sortedWordCount => println(sortedWordCount.\_1 + " appear " + sortedWordCount.\_2 + "times"))

**掌握简单排序、了解二次排序**

2.1在/simple/目录下使用命令:touch sort.txt创建名为sort.txt数据文本，数据信息如下：

2 3

3 4

2 1

3 3

1 5

1 4

2.2编辑上面创建的对象“SecondSortKey”，然后编辑内容如下：

**class** SecondSortKey(**val** first: Int, **val** second: Int) **extends** *Ordered*\[SecondSortKey\] **with** Serializable \{

**def** compare(that: SecondSortKey): Int = \{

**if**(**this**.first - that.first != 0) \{

**this**.first - that.first

\} **else** \{

**this**.second - that.second

2.3 编辑上面创建的对象“SecondSort”，然后编辑内容如下：

// 二次排序

// 1、实现自定义的key，要实现Ordered接口和Serializable接口，在key中实现

// 己对多个列的排序算法。使用sortByKey算子按照自定义的key进行排序

**import** org.apache.spark.SparkConf

**import** org.apache.spark.SparkContext

**object** SecondSort \{

**def** main(args: Array\[*String*\]): Unit = \{

// 创建SparkConf

**val** conf = **new** SparkConf()

.setAppName("SecondSort")

.setMaster("local")

// 创建SparkContext

**val** sc = **new** SparkContext(conf)

// 读取文件

**val** lines = sc.textFile("/simple/sort.txt", 1)

// 对获得的元素按空格分割并调用SecondSortKey函数

**val** pairs = lines.map \{ line => (

**new** SecondSortKey(line.split(" ")(0).toInt, line.split(" ")(1).toInt),

line)\}

// 按降序排序

**val** sortedPairs = pairs.sortByKey()

// 获取sortedPair数组中第二个值

**val** sortedLines = sortedPairs.map(sortedPair => sortedPair.\_2)

// 遍历打印输出

sortedLines.foreach \{ sortedLine => println(sortedLine) \}

**掌握抓取文本中最大的前几位数字、掌握排序算子的使用**

2.1在/simple/目录下使用命令:touch top.txt创建名为top.txt数据文本，数据信息如下：

234

2.2编辑上面创建获取前几个数字的对象“Top3”，然后编辑内容如下：

**object** Top3 \{

**def** main(args: Array\[*String*\])\{

// 创建SparkConf

**val** conf = **new** SparkConf().setAppName("Top3").setMaster("local")

// 创建SparkContext

**val** sc = **new** SparkContext(conf)

// 读取文件

**val** lines = sc.textFile("/simple/top.txt", 1)

// 对linesRDD进行map操作

**val** pairs = lines.map \{ line => (line.toInt,line) \}

// 对pairsRDD进行降序sortByKey操作

**val** sortedPairs = pairs.sortByKey(**false**)

// 对降序的RDD再次进行map操作，获取每一行的第一个数

**val** sortedNumbers = sortedPairs.map(sortedPair => sortedPair.\_1)

// 取前3个数

**val** top3Number = sortedNumbers.take(3)

// 打印输出

**for**(num <- top3Number)\{

println(num)

**掌握获取不同类中前几部分的操作、掌握分组算子的使用**

2.1在/simple/目录下使用命令:touch classsort.txt创建名为classsort.txt数据文本：

class1 90

class2 56

class3 87

class1 76

class2 88

class1 95

class1 74

class2 87

class2 67

class2 77

2.2编辑上面创建的对象“GroupTop3，然后编辑内容如下：

**object** GroupTop3 \{

**def** main(args: Array\[*String*\])\{

// 创建SparkConf

**val** conf = **new** SparkConf().setAppName("Top3").setMaster("local")

// 创建SparkContext

**val** sc = **new** SparkContext(conf)

// 读取文件

**val** lines = sc.textFile("/simple/classsort.txt", 1)

// 对获取的元素进行按空格切割成两部分

**val** pairs=lines.map\{line=>(line.split(" ")(0),line.split(" ")(1).toInt)\}

// 对pairs中的每个元素进行按键分组操作

**val** grouped=pairs.groupByKey

// 对分组的元素按分数取前3

**val** groupedTop3=grouped.map(grouped=>

\{(grouped.\_1,grouped.\_2.toList.sortWith(\_>\_).take(3))\})

// 进行降序排列（默认是降序）

**val** groupedKeySorted=groupedTop3.sortByKey()

// 打印输出

groupedKeySorted.foreach(pair=>\{

println(pair.\_1+":")

pair.\_2.foreach \{println\}

\})

**Spark Streaming****实时计算**

Spark Streaming内部的基本工作原理如下：接收实时输入数据流，然后将数据拆分成多个batch，比如每收集1秒的数据封装为一个batch，然后将每个batch交给Spark的计算引擎进行处理，最后会生产出一个结果数据流，其中的数据，也是由一个一个的batch所组成的

![image002.jpg][]![uploading.4e448015.gif][]转存失败重新上传取消![http://10.36.52.1/upload/init/initimage/1511489461608/image002.jpg][image002.jpg]

Spark Streaming提供了一种高级的抽象，叫做DStream，英文全称为Discretized Stream，中文翻译为“离散流”，它代表了一个持续不断的数据流。DStream可以通过输入数据源来创建，比如Kafka、Flume和Kinesis；也可以通过对其他DStream应用高阶函数来创建，比如map、reduce、join、window。

DStream的内部，其实一系列持续不断产生的RDD。RDD是Spark Core的核心抽象，即，不可变的，分布式的数据集。DStream中的每个RDD都包含了一个时间段内的数据。

![image004.jpg][]![uploading.4e448015.gif][]转存失败重新上传取消![http://10.36.52.1/upload/init/initimage/1511489461608/image004.jpg][image004.jpg]

对DStream应用的算子，比如map，其实在底层会被翻译为对DStream中每个RDD的操作。比如对一个DStream执行一个map操作，会产生一个新的DStream。但是，在底层，其实其原理为，对输入DStream中每个时间段的RDD，都应用一遍map操作，然后生成的新的RDD，即作为新的DStream中的那个时间段的一个RDD。底层的RDD的transformation操作，其实，还是由Spark Core的计算引擎来实现的。Spark Streaming对Spark Core进行了一层封装，隐藏了细节，然后对开发人员提供了方便易用的高层次的API。

![image006.jpg][]![uploading.4e448015.gif][]转存失败重新上传取消![http://10.36.52.1/upload/init/initimage/1511489461608/image006.jpg][image006.jpg]

2.1安装nc工具：yum install nc

2.2编辑上面创建的对象“WordCount”，然后编辑内容如下：

**import** org.apache.spark.SparkConf

**import** org.apache.spark.streaming.Seconds

**import** org.apache.spark.streaming.StreamingContext

**object** WordCount \{

**def** main(args: Array\[*String*\]): Unit = \{

// 创建SparkConf

**val** conf = **new** SparkConf()

.setMaster("local\[2\]")

.setAppName("WordCount")

// 创建StreamingContext并设置间隔为20秒

**val** ssc = **new** StreamingContext(conf, Seconds(20))

// 监听本机9999端口

**val** lines = ssc.socketTextStream("localhost", 9999)

// 对linesRDD进行按空格切割操作

**val** words = lines.flatMap \{ \_.split(" ") \}

// 将分割的单词后面都加上一个1，组成键值对

**val** pairs = words.map \{ word => (word, 1) \}

// 队键值对进行按键相加操作

**val** wordCounts = pairs.reduceByKey(\_ + \_)

// 打印输出结果

wordCounts.print()

// 关闭StreamingContext

ssc.start()

ssc.awaitTermination()

3.1 在命令行界面执行nc \-lk 9999

3.2 选中对象类”WordCount“，右键点击选择”Run as”--->”Scala Application”,监控9999端口

**Spark****创建****RDD****的三种方式**

1.从scala集合中创建RDD，通过parallelize方法创建RDD.

2.通过读取本地文件，HDFS，DB等数据源，使用方法，如textFile方法，创建RDD。

3.通过从其他的RDD转换成新RDD。

2.1 编辑上面创建的对象“MakeRDDType”，然后编辑内容如下：

package com.simple

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object MakeRDDType \{

def main(args: Array\[String\])\{

val conf = new SparkConf().setAppName("MakeRDDTest").setMaster("local\[4\]")

val sc = new SparkContext(conf)

//从scala集合中创建RDD

val nums = 1 to 100

val rdd = sc.parallelize(nums)

val sum = rdd.reduce(\_+\_)

println("sum:"+sum)

//从本地文件系统创建RDD

val data = sc.textFile("/simple/spark-1.6.1-bin-hadoop2.4/README.md")

val wordCounts = data.flatMap(line => line.split(" ")).map(word =>(word,1)).reduceByKey(\_+\_)

wordCounts.foreach(println)

//从其他RDD转换成RDD

val rdd1 = sc.textFile("/simple/spark-1.6.1-bin-hadoop2.4/README.md")

val linesLength =rdd1.map \{ line => line.length \}

//在lingsLength RDD基础上进行相加

val sum1=linesLength.reduce(\_+\_)

print("the total length of the file of ReadMe："+sum1)

**熟悉****Join,LeftOuterJoin,RightOuterJoin,Union****原理和操作**

Join, 首先，第一步先对两个RDD进行cogroup操作，也就是把每个RDD中的元素构成集合，然后再合并成一个集合，第二步是：进行笛卡尔积.

def join\[W\](other: RDD\[(K, W)\]): RDD\[(K, (V, W))\]  
def join\[W\](other: RDD\[(K, W)\], numPartitions: Int): RDD\[(K, (V, W))\]  
def join\[W\](other: RDD\[(K, W)\], partitioner: Partitioner): RDD\[(K, (V, W))\]

LeftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。

def leftOuterJoin\[W\](other: RDD\[(K, W)\]): RDD\[(K, (V, Option\[W\]))\]  
def leftOuterJoin\[W\](other: RDD\[(K, W)\], numPartitions: Int): RDD\[(K, (V, Option\[W\]))\]  
def leftOuterJoin\[W\](other: RDD\[(K, W)\], partitioner: Partitioner): RDD\[(K, (V, Option\[W\]))\]

RightOuterJoin类似于SQL中的有外关联right outer join，返回结果以参数中的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可

def rightOuterJoin\[W\](other: RDD\[(K, W)\]): RDD\[(K, (Option\[V\], W))\]  
def rightOuterJoin\[W\](other: RDD\[(K, W)\], numPartitions: Int): RDD\[(K, (Option\[V\], W))\]  
def rightOuterJoin\[W\](other: RDD\[(K, W)\], partitioner: Partitioner): RDD\[(K, (Option\[V\], W))\]

Union,把两个RDD联合起来，两个RDD里面的数据类型一定要相同

def union(other: RDD\[T\]): RDD\[T\]

2.1 编辑上面创建的对象“JoinTest”，然后编辑内容如下：

package com.simple

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object JoinTest \{

def main(args: Array\[String\]) \{

val conf = new SparkConf().setAppName("JoinTest").setMaster("local\[4\]")

val sc = new SparkContext(conf)

val a = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)

val bKey = a.keyBy(\_.length)

val c = sc.parallelize(List("dog", "cat", "gnu", "salmon", "rabbit", "turkey", "wolf", "bear", "bee"), 3)

val dKey = c.keyBy(\_.length)

println("-----bKey data-----")

bKey.collect().foreach(line => print(line + " "))

println()

println("-----dKey data-----")

dKey.collect().foreach(line => print(line + " "))

//join操作

bKey.join(dKey).collect().foreach(line => print(line+ " "))

//leftOuterJoin操作

bKey.leftOuterJoin(dKey).collect().foreach(line => print(line + " "))

//rightOuterJoin操作

bKey.rightOuterJoin(dKey).collect().foreach(line => print(line + " "))

//union操作

bKey.union(dKey)

**map****和****flatmap**

Spark支持两种RDD操作：transformation和action。transformation操作会针对已有的RDD创建一个新的RDD；而action则主要是对RDD进行最后的操作，比如遍历、reduce、保存到文件等，并可以返回结果给Driver程序。例如，map就是一种transformation操作，它用于将已有RDD的每个元素传入一个自定义的函数，并获取一个新的元素，然后将所有的新元素组成一个新的RDD。而reduce就是一种action操作，它用于对RDD中的所有元素进行聚合操作，并获取一个最终的结果，然后返回给Driver程序。

transformation的特点就是lazy特性。lazy特性指的是，如果一个spark应用中只定义了transformation操作，那么即使你执行该应用，这些操作也不会执行。也就是说，transformation是不会触发spark程序的执行的，它们只是记录了对RDD所做的操作，但是不会自发的执行。只有当transformation之后，接着执行了一个action操作，那么所有的transformation才会执行。Spark通过这种lazy特性，来进行底层的spark应用执行的优化，避免产生过多中间结果。action操作执行，会触发一个spark job的运行，从而触发这个action之前所有的transformation的执行。这是action的特性。

1.1执行命令：start-all.sh启动hadoop服务

1.2进入spark的安装目录/simple/spark-1.6.1bin-hadoop2.4/sbin文件夹下执行命令：./start-all.sh启动spark服务

1.3使用jps命令查看启动的所有服务

2.1进入spark安装目录/simple/spark-1.6.1bin-hadoop2.4/bin文件夹下执行spark-shell命令，启动spark shell界面

2.3输入代码进行map算子操作

**val** numbers = Array(1,2,3,4,5)

**val** numberRDD = sc.parallelize(numbers, 1)

**val** multipleNumberRDD = numberRDD.map \{ num=> num \* 2 \}

multipleNumberRDD.foreach \{ num=>println(num)\}

2.4输入代码进行flatmap算子操作

**val** lineArray = Array("hello you","hello me","hello world")

**val** lines = sc.parallelize(lineArray, 1)

**val** words = lines.flatMap \{ line => line.split(" ") \}

words.foreach \{ word => println(word) \}

**JSON****数据的读取方法**

Spark 1.0版本开始，推出了Spark SQL。其实最早使用的，都是Hadoop自己的Hive查询引擎；但是后来Spark提供了Shark；再后来Shark被淘汰，推出了Spark SQL。Shark的性能比Hive就要高出一个数量级，而Spark SQL的性能又比Shark高出一个数量级。

最早来说，Hive的诞生，主要是因为要让那些不熟悉Java工程师，无法深入进行MapReduce编程的数据分析师，能够使用他们熟悉的关系型数据库的SQL模型，来操作HDFS上的数据。因此推出了Hive。Hive底层基于MapReduce实现SQL功能，能够让数据分析人员，以及数据开发人员，方便的使用Hive进行数据仓库的建模和建设，然后使用SQL模型针对数据仓库中的数据进行统计和分析。但是Hive有个致命的缺陷，就是它的底层基于MapReduce，而MapReduce的shuffle又是基于磁盘的，因此导致Hive的性能异常低下。进而出现复杂的SQL ETL，要运行数个小时，甚至数十个小时的情况。

后来，Spark推出了Shark，Shark与Hive实际上还是紧密关联的，Shark底层很多东西还是依赖于Hive，但是修改了内存管理、物理计划、执行三个模块，底层使用Spark的基于内存的计算模型，从而让性能比Hive提升了数倍到上百倍。

然而，Shark还是它的问题所在，Shark底层依赖了Hive的语法解析器、查询优化器等组件，因此对于其性能的提升还是造成了制约。所以后来Spark团队决定，完全抛弃Shark，推出了全新的Spark SQL项目。Spark SQL就不只是针对Hive中的数据了，而且可以支持其他很多数据源的查询。

Spark SQL的特点

1、支持多种数据源：Hive、RDD、Parquet、JSON、JDBC等。

2、多种性能优化技术：in-memory columnar storage、byte-code generation、cost model动态评估等。

3、组件扩展性：对于SQL的语法解析器、分析器以及优化器，用户都可以自己重新开发，并且动态扩展。

2.1在/simple/目录下使用命令:touch students.json编写名为students.json的JSON数据源

**\{"id":1, "name":"leo", "age":18\}**

**\{"id":2, "name":"jack", "age":19\}**

**\{"id":3, "name":"marry", "age":17\}**

2.2编辑上面创建的对象“DataFrameCreate”，然后编辑内容如下：

**object** DataFrameCreate \{

**def** main(args: Array\[*String*\])\{

// 创建SparkConf

**val** conf = **new** SparkConf().setAppName("DataFrameCreate").setMaster("local")

// 创建SparkContext

**val** sc = **new** SparkContext(conf)

// 创建SQLContext

**val** sqlContext = **new** SQLContext(sc)

// 读取json格式的数据

**val** df = sqlContext.read.json("/simple/students.json")

// 展示输出

df.show()

**json****综合分析**

【实验目的】

1、掌握DataFrame的使用 2、了解SparkSQL的知识点

【实验原理】

为什么要将RDD转换为DataFrame？因为这样的话，我们就可以直接针对HDFS等任何可以构建为RDD的数据，使用Spark SQL进行SQL查询了。这个功能是无比强大的。想象一下，针对HDFS中的数据，直接就可以使用SQL进行查询。

Spark SQL支持两种方式来将RDD转换为DataFrame。

第一种方式，是使用反射来推断包含了特定数据类型的RDD的元数据。这种基于反射的方式，代码比较简洁，当你已经知道你的RDD的元数据时，是一种非常不错的方式。

第二种方式，是通过编程接口来创建DataFrame，你可以在程序运行时动态构建一份元数据，然后将其应用到已经存在的RDD上。这种方式的代码比较冗长，但是如果在编写程序时，还不知道RDD的元数据，只有在程序运行时，才能动态得知其元数据，那么只能通过这种动态构建元数据的方式。

2.1在/simple/目录下使用命令:touch student.json创建名为student.json数据文本

\{"name":"Leo", "score":85\}

\{"name":"Marry", "score":99\}

\{"name":"Jack", "score":74\}

2.2编辑上面创建的对象“JSONDataSource"，然后编辑内容如下

**import** org.apache.spark.SparkConf

**import** org.apache.spark.SparkContext

**import** org.apache.spark.sql.SQLContext

**import** org.apache.spark.sql.types.**StructType**

**import** org.apache.spark.sql.types.**StructField**

**import** org.apache.spark.sql.types.StringType

**import** org.apache.spark.sql.types.IntegerType

**import** org.apache.spark.sql.Row

**import** org.apache.spark.sql.types.LongType

**object** JSONDataSource \{

**def** main(args: Array\[*String*\]): Unit = \{

**val** conf = **new** SparkConf()

.setAppName("JSONDataSource").setMaster("local")

**val** sc = **new** SparkContext(conf)

**val** sqlContext = **new** SQLContext(sc)

// 创建学生成绩DataFrame

**val** studentScoresDF = sqlContext.read.json("/simple/students.json")

// 查询出分数大于80分的学生成绩信息，以及学生姓名

studentScoresDF.registerTempTable("student\_scores")

**val** goodStudentScoresDF = sqlContext.sql("select name,score from student\_scores where score>=80")

**val** goodStudentNames = goodStudentScoresDF.rdd.map \{ row => row(0) \}.collect()

// 创建学生基本信息DataFrame

**val** studentInfoJSONs = Array("\{\\"name\\":\\"Leo\\", \\"age\\":18\}",

"\{\\"name\\":\\"Marry\\", \\"age\\":17\}",

"\{\\"name\\":\\"Jack\\", \\"age\\":19\}")

**val** studentInfoJSONsRDD = sc.parallelize(studentInfoJSONs, 3);

**val** studentInfosDF = sqlContext.read.json(studentInfoJSONsRDD)

// 查询分数大于80分的学生的基本信息

studentInfosDF.registerTempTable("student\_infos")

**var** sql = "select name,age from student\_infos where name in ("

**for**(i <- 0 until goodStudentNames.length) \{

sql += "'" + goodStudentNames(i) + "'"

**if**(i < goodStudentNames.length - 1) \{

sql += ","

sql += ")"

**val** goodStudentInfosDF = sqlContext.sql(sql)

// 将分数大于80分的学生的成绩信息与基本信息进行join

**val** goodStudentsRDD =

goodStudentScoresDF.rdd.map \{ row => (row.getAs\[*String*\]("name"), row.getAs\[Long\]("score")) \}

.join(goodStudentInfosDF.rdd.map \{ row => (row.getAs\[*String*\]("name"), row.getAs\[Long\]("age")) \})

// 将rdd转换为dataframe

**val** goodStudentRowsRDD = goodStudentsRDD.map(

info => Row(info.\_1, info.\_2.\_1.toInt, info.\_2.\_2.toInt))

**val** structType = **StructType**(Array(

**StructField**("name", **StringType**, **true**),

**StructField**("score", **IntegerType**, **true**),

**StructField**("age", **IntegerType**, **true**)))

**val** goodStudentsDF = sqlContext.createDataFrame(goodStudentRowsRDD, structType)

// 将dataframe中的数据保存到json中

goodStudentsDF.write.format("json").save("/simple/good-students-JSON")

**熟悉****filter,distinct,Cartesian****原理和操作**

Filter是一个过滤操作，每一个V的值都要经过f函数，满足的会留下，产生最终结果。

Distinct作用是，去除RDD中相同的元素

Cartesian就是对给的两个RDD进行笛卡儿计算

2.1 编辑上面创建的对象“FilterDistRDDTest”

package com.simple

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object FilterDistRDDTest \{

def main(args: Array\[String\])\{

val conf = new SparkConf().setAppName("FilterDistRDDTest").setMaster("local\[4\]")

val sc = new SparkContext(conf)

val a = sc.parallelize(1 to 10, 3)

//filter操作

val filterData = a.filter(\_ % 2 == 0)

filterData.collect.foreach(line => print(line + " "))

//distinct操作

val dist = sc.parallelize(List("Gnu", "Cat", "Rat", "Dog", "Gnu", "Rat"), 2)

dist.distinct.collect().foreach(line => print(line + " "))

//cartesian操作

val x = sc.parallelize(List(1,2,3,4,5))

val y = sc.parallelize(List(6,7,8,9,10))

x.cartesian(y).collect.foreach(line => print(line + " "))

[image002.jpg]: http://10.36.52.1/upload/init/initimage/1511489461608/image002.jpg
[uploading.4e448015.gif]: https://csdnimg.cn/release/mp/img/uploading.4e448015.gif
[image004.jpg]: http://10.36.52.1/upload/init/initimage/1511489461608/image004.jpg
[image006.jpg]: http://10.36.52.1/upload/init/initimage/1511489461608/image006.jpg