Spark大数据分析与实战笔记（第二章 Spark基础-06）

文章目录

每日一句正能量
- 2.6 IDEA开发WordCount程序
- - 2.6.1 本地模式执行Spark程序
  - 2.6.2 集群模式执行Spark程序

每日一句正能量

我们全都要从前辈和同辈学习到一些东西。就连最大的天才，如果想单凭他所特有的内在自我去对付一切，他也决不会有多大成就。

2.6 IDEA开发WordCount程序

Spark-Shell通常在测试和验证我们的程序时使用的较多，然而在生产环境中,通常会在IDEA开发工具中编写程序,然后打成Jar包，最后提交到集群中执行。本节我们将利用IDEA工具开发一个WordCount单词计数程序。

2.6.1 本地模式执行Spark程序

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

创建Maven项目,新建资源文件夹创建Maven项目,新建资源文件夹

创建一个名为“spark_chapter02”的Maven项目
然后在main和test目录下分别创建一个名称为scala的文件夹。
将mian下的scala文件夹标记为【资源文件夹】（【右键】→【Mak Directory as】→【Sources Root】）。
将test下的scala文件夹标记为【测试的资源文件夹】（【右键】→【Mak Directory as】→【TestSourcesRoot】）。

如下图所示：

在这里插入图片描述

添加Spark相关依赖，打包插件

设置依赖的版本号
添加了Scala、Hadoop和Spark相关的依赖

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

	
    
        2.11.8
        2.7.4
        2.3.2
    
    
        
        
            org.scala-lang
            scala-library
            ${scala.version}
        
        
        
            org.apache.spark
            spark-core_2.11
            ${spark.version}
        
        
        
            org.apache.hadoop
            hadoop-client
            ${hadoop.version}

编写代码，查看结果

在main目录的scala文件夹中，创建WordCount.scala文件实现词频统计。创建WordCount.scala，选择Object，如下图所示：

WordCount.scala代码如下所示：

package cn.itcastimport org.apache.spark.rdd.RDDimport org.apache.spark.{SparkConf, SparkContext}object WordCount {  def main(args: Array[String]): Unit = {    //1:创建SparkConf对象，设置appName和Master地址val sparkConf = new SparkConf().setAppName("Wordcount").setMaster("local[2]")    //2:创建SparkContext对象，他是所有任务计算的源头    val sparkContext = new SparkContext(sparkConf)    //3:读取数据文件,RDD简单理解为一个集合    val data: RDD[String] = sparkContext.textFile("c:\\word\\words.txt")    //4:切分乱每一行，获取所有的单词，单词之间以空格隔开    val words: RDD[String] = data.flatMap(_.split(" "))    //5:每个单词标记为1，转换为（单词，1）    val wordAndOne: RDD[(String, Int)] = words.map(x => (x, 1))    //6:相同的单词汇总,前一个下划线表示累加数据，后一个下划线表示新数据 (x,y)=>x+y    val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)    //7:收集打印结果数据    val finalResult: Array[(String, Int)] = result.collect()    println(finalResult.toBuffer)    //8:关闭sparkContext对象    sparkContext.stop()  }}

文件 c:\word\words.txt内容如下：

hello hadoop
hello spark
hello itcast

运行结果如下所示：

在这里插入图片描述

2.6.2 集群模式执行Spark程序

集群模式是指将Spark程序提交至Spark集群中执行任务，由Spark集群负责资源的调度，程序会被框架分发到集群中的节点上并发地执行。下面分步骤介绍如何在集群模式下执行Spark程序。

添加打包插件

在实际工作应用中，代码编写完成后，需要将程序打包，上传至服务器运行，因此还需要向pom.xml文件中添加所需插件，具体配置参数如下。

        src/main/scala        src/test/scala                                    net.alchim31.maven                scala-maven-plugin                3.2.2                                                                                        compile                            testCompile                                                                                                            -dependencyfile                                ${project.build.directory}/.scala_dependencies                                                                                                                                org.apache.maven.plugins                maven-shade-plugin                2.4.3                                                            package                                                    shade                                                                                                                                                *:*                                                                            META-INF/*.SF                                        META-INF/*.DSA                                        META-INF/*.RSA

注意：如果在创建Maven工程中选择Scala原型模板，上述插件会自动创建。这些插件的主要功能是方便开发人员进行打包。

修改代码，打包程序

在打包项目之前，需要对词频统计的代码进行修改，创建WordCount_Online.scala文件，代码如下所示。

package cn.itcast

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

//编写单词计数程序，打成Jar包，提交到集群中运行
object WordCount_Online {
  def main(args: Array[String]): Unit = {
    //1:创建SparkConf对象，设置appName和Master地址
    val sparkConf = new SparkConf().setAppName("Wordcount_Online")

    //2:创建SparkContext对象，他是所有任务计算的源头
    val sparkContext = new SparkContext(sparkConf)
    //3:读取数据文件(path是HDFS上的路径，以传参的方式),RDD简单理解为一个集合
    val data: RDD[String] = sparkContext.textFile(args(0))
    //4:切分乱每一行，获取所有的单词，单词之间以空格隔开
    val words: RDD[String] = data.flatMap(_.split(" "))
    //5:每个单词标记为1，转换为（单词，1）
    val wordAndOne: RDD[(String, Int)] = words.map(x => (x, 1))
    //6:相同的单词汇总,前一个下划线表示累加数据，后一个下划线表示新数据 (x,y)=>x+y
    val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
    //7:将结果数据保存到HDFS上
    result.saveAsTextFile(args(1))

    //8:关闭sparkContext对象
    sparkContext.stop()
  }
}

将程序打包

打好的jar包在target中。项目生成了两个Jar包，其中original包中不含有第三方Jar包：spark_chapter02-1.0-SNAPSHOT.jar包。如下所示：

将spark_chapter02-1.0-SNAPSHOT.jar包上传至hadoop01节点中的/export/data路径下。如下图所示：

执行提交命令“spark-submit”

先启动Spark。如下图所示
在hadoop01节点的spark目录下，执行“spark-submit” 命令提交任务，命令如下。

bin/spark-submit --master spark://hadoop01:7077 \--class cn.itcast.WordCount_Online \--executor-memory 1g \--total-executor-cores 1 \/export/data/spark_chapter02-1.0-SNAPSHOT.jar \/spark/test/words.txt \/spark/test/out

结果如下图所示：

在这里插入图片描述