Spark基础学习笔记（1）

一、案例分析：Spark RDD实现单词计数
（一）案例概述
单词计数是学习分布式计算的入门程序，有很多种实现方式，例如MapReduce；使用Spark提供的RDD算子可以更加轻松地实现单词计数。
在IntelliJ IDEA中新建Maven管理的Spark项目，并在该项目中使用Scala语言编写Spark的WordCount程序，最后将项目打包提交到Spark集群（Standalone模式）中运行。
（二）实现步骤
1、新建Maven管理的Spark项目
在IDEA中选择File→new→Project…，在弹出的窗口中选择左侧的Maven项，然后在右侧勾选Create fromarchetype复选框并选择下方出现的org.scala-tools.archetypes:scala-archetype-simple项（表示使用scala-archetype-simple模板构建Maven项目）。

在弹出的窗口中填写GroupId与ArtifactId，Version保持默认设置即可，然后单击Next按钮

在弹出的窗口中从本地系统选择Maven安装的主目录的路径、Maven的配置文件settings.xml的路径以及Maven仓库的路径，然后单击Next按钮

在弹出的窗口中项目名称为WordCount，就是先前设置的ArtifactId的值，当然也可以修改，然后单击Finish按钮

2、添加Scala和Spark依赖

启动spark-shell，可以看到Spark2.4.4使用Scala2.11.12

在pom.xml文件里添加Scala2.11.12和Spark 2.4.4依赖，添加Maven构建插件
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>net.hw.spark</groupId>
<artifactId>WordCount</artifactId>
<version>1.0-SNAPSHOT</version>
<inceptionYear>2008</inceptionYear>

<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.12</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>2.4.4</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
<archive>
<manifest>

<mainClass>net.hw.spark.WordCount</mainClass>
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.3.2</version>
<executions>
<execution>
<id>scala-compile-first</id>
<phase>process-resources</phase>
<goals>
<goal>add-source</goal>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>scala-test-compile</id>
<phase>process-test-resources</phase>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

3、创建WordCount对象
在net.hw.spark包里创建WordCount对象

package net.hw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
* 功能：统计单词个数

*/
object WordCount {
def main(args: Array[String]): Unit = {
// 创建SparkConf对象，存储应用程序配置信息
val conf = new SparkConf()
.setAppName("Spark-WordCount") // 设置应用程序名称，可在Spark WebUI中显示
.setMaster("spark://master:7077") // 设置集群Master节点访问地址
// 创建SparkContext对象，该对象是提交Spark应用程序的入口
val sc = new SparkContext(conf)
// 读取指定路径（程序执行时传入的第一个参数）的文件内容，生成一个RDD
val rdd: RDD[String] = sc.textFile(args(0))
// 对rdd进行处理
rdd.flatMap(_.split(" ")) // 将RDD的每个元素按照空格进行拆分并将结果合并为一个新RDD
.map((_, 1)) //将RDD中的每个单词和数字1放到一个元组里，即(word,1)
.reduceByKey(_ + _) //对单词根据key进行聚合，对相同的key进行value的累加
.sortBy(_._2, false) // 按照单词数量降序排列
.saveAsTextFile(args(1)) //保存结果到指定的路径(取程序执行时传入的第二个参数)
//停止SparkContext，结束该任务
sc.stop();
}
}
4、对于程序代码进行解析
SparkConf对象的setMaster()方法用于设置Spark应用程序提交的URL地址。若是Standalone集群模式，则指Master节点的访问地址；若是本地（单机）模式，则需要将地址改为local或local[N]或local[*]，分别指使用1个、N个和多个CPU核心数。本地模式可以直接在IDE中运行程序，不需要Spark集群。
此处也可不设置。若将其省略，则使用spark-submit提交该程序到集群时必须使用--master参数进行指定。
SparkContext对象用于初始化Spark应用程序运行所需要的核心组件，是整个Spark应用程序中很重要的一个对象。启动Spark Shell后默认创建的名为sc的对象即为该对象。
textFile()方法需要传入数据来源的路径。数据来源可以是外部的数据源（HDFS、S3等），也可以是本地文件系统（Windows或Linux系统），路径可以使用以下3种方式：
（1）文件路径：例如textFile("/input/data.txt ")，此时将只读取指定的文件。
（2）目录路径：例如textFile("/input/words/")，此时将读取指定目录words下的所有文件，不包括子目录。
（3）路径包含通配符：例如textFile("/input/words/*.txt")，此时将读取words目录下的所有TXT文件。
该方法将读取的文件中的内容按行进行拆分并组成一个RDD集合。假设读取的文件为words.txt，则上述代码的具体数据转化流程如下图所示。

5、将Spark项目编译和打包

展开IDEA右侧的Maven Projects窗口，双击其中的package项，将编写好的WordCount项目进行编译和打包

生成两个jar包，一个没有带依赖，一个带了依赖，我们使用没有带依赖的jar包