本地模式执行Spark程序 - 大数据学习教程_大数据培训_ 大数据技术与应用

Spark作业与MapReduce作业同样可以先在本地开发测试，本地执行模式与集群提交模式，代码的业务功能相同，因此本书大多数采用本地开发模式。下面讲解使用IDEA工具开发WordCount单词计数程序的相关步骤。

1．创建Maven项目，新建资源文件夹

创建一个Maven工程项目，名为“spark_chapter02”。项目创建好后，在main和test目录下分别创建一个名称为scala的文件夹，创建好的目录结构如图1所示。

图1 Spark_chapter02项目目录结构

在图1中，选中main目录下的scala文件夹，单击右键选择【Mark Directory as】→【Sources Root】，将文件夹标记为资源文件夹类型；同样的，选中test目录下的scala文件夹，单击右键选择【Mark Directory as】→【Test Sources Root】将文件夹标记为测试资源文件夹类型。其中，资源文件夹中存放项目源码文件，测试文件夹中存放开发中测试的源码文件。

2．添加Spark相关依赖、打包插件

Maven是一个项目管理工具，虽然我们刚才创建好了项目，但是却不能识别Spark类，因此，我们需要将Spark相关的依赖添加到Maven项目中。打开pom.xml文件，在该文件中添加的依赖如下所示：

 1  <!--设置依赖版本号-->
 2    <properties>
 3      <scala.version>2.11.8</scala.version>
 4      <hadoop.version>2.7.4</hadoop.version>
 5      <spark.version>2.3.2</spark.version>
 6    </properties>
 7    <dependencies>
 8      <!--Scala-->
 9      <dependency>
 10       <groupId>org.scala-lang</groupId>
 11       <artifactId>scala-library</artifactId>
 12       <version>${scala.version}</version>
 13     </dependency>
 14     <!--Spark-->
 15     <dependency>
 16       <groupId>org.apache.spark</groupId>
 17       <artifactId>spark-core_2.11</artifactId>
 18       <version>${spark.version}</version>
 19     </dependency>
 20     <!--Hadoop-->
 21     <dependency>
 22       <groupId>org.apache.hadoop</groupId>
 23       <artifactId>hadoop-client</artifactId>
 24       <version>${hadoop.version}</version>
 25     </dependency>
 26   </dependencies>

在上述配置参数片段中，<properties>标签用来设置所需依赖的版本号，其中在<dependencies>标签中添加了Scala、Hadoop和Spark相关的依赖，设置完毕后，相关Jar文件会被自动加载到项目中。

3．编写代码，查看结果

在main目录下的scala文件夹中，创建WordCount.scala文件用于词频统计，代码如文件1所示。

文件1 WordCount.scala

 1  import org.apache.spark.rdd.RDD
 2  import org.apache.spark.{SparkConf, SparkContext}
 3  //编写单词计数
 4  object WordCount {
 5    def main(args: Array[String]): Unit = {
 6      //1.创建SparkConf对象,设置appName和Master地址
 7      val sparkconf = new 
 8         SparkConf().setAppName("WordCount").setMaster("local[2]")
 9      //2.创建SparkContext对象，它是所有任务计算的源头，
 10     // 它会创建DAGScheduler和TaskScheduler
 11     val sparkContext = new SparkContext(sparkconf)
 12     //3.读取数据文件,RDD可以简单的理解为是一个集合
 13     // 集合中存放的元素是String类型
 14     val data : RDD[String] = 
 15                  sparkContext.textFile("D:\\word\\words.txt")
 16     //4.切分每一行，获取所有的单词
 17     val words :RDD[String] = data.flatMap(_.split(" "))
 18     //5.每个单词记为1，转换为（单词，1）
 19     val wordAndOne :RDD[(String, Int)] = words.map(x =>(x,1))
 20     //6.相同单词汇总，前一个下划线表示累加数据，后一个下划线表示新数据
 21     val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)
 22     //7.收集打印结果数据
 23     val finalResult: Array[(String, Int)] = result.collect()
 24     println(finalResult.toBuffer)
 25     //8.关闭sparkContext对象
 26     sparkContext.stop()
 27   }
 28 }

上述代码中，第7-11行代码创建SparkContext对象并通过SparkConf对象设置配置参数，其中Master为本地模式，即可以在本地直接运行；第14-24行代码中，读取数据文件，将获得的数据按照空格切分，将每个单词记作（单词，1），之后若出现相同的单词就将次数累加，最终打印数据结果；第26行代码表示关闭SparkContext对象资源。执行代码成功后，在控制台可以查看输出结果，如图2所示。

图2 IDEA开发WordCount

从图2可以看出，文本中的单词已经成功统计了出现的次数。