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单词计数程序的相关步骤。

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

        如下图所示：

        1. 添加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}        

          1. 编写代码，查看结果
          • 在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 hadoophello sparkhello 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原型模板，上述插件会自动创建。这些插件的主要功能是方便开发人员进行打包。

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

                package cn.itcastimport org.apache.spark.rdd.RDDimport 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()  }}

                1. 将程序打包

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

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

                • 执行提交命令“spark-submit”
                  1. 先启动Spark。如下图所示
                  2. 在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

                  结果如下图所示：

                  1. 验证结果，如下图所示：

                  转载自：https://blog.csdn.net/u014727709/article/details/132509490

                  欢迎 点赞✍评论⭐收藏，欢迎指正