Spark实战演练：WordCount实例

╰+攻爆jí腚メ 2021-09-11 01:22 488阅读 0赞

摘要：本文主要讲了如何使用Spark来运行一个wordCount实例

1、本地运行实例

package com.lin.wordcount
    import org.apache.spark.SparkConf
    import org.apache.spark.SparkContext
    import org.apache.spark.SparkContext._
    
    /**
     * 本地运行Spark
     * 环境：Windows7
     * scala版本：2.11.8
     * Spark版本：1.6.1
     */
    object WordCountLocal {
    
      def main(args: Array[String]): Unit = {
        /**
         * 第一步:创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息
         * 例如说通过setMaster来设置程序要连接的Spark集群的Master的URL
         * 如果设置为local，则代表Spark程序在本地运行，特别适合于配置条件的较差的人
         *
         */
    
        val conf = new SparkConf()
        conf.setAppName("wordCountLocal") //设置应用程序的名称，在程序运行的监控界面可以看到名称  
        conf.setMaster("local") //此时程序在本地运行，无需安装Spark的任何集群  
    
        /**
         * 第二步:创建SparkContext对象
         * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala，Java，Python等都必须有一个SparkContext
         * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler，TaskScheduler，Scheduler
         * 同时还会负责Spark程序往Master注册程序等
         * SparkContext是整个Spark应用程序中最为至关重要的一个对象。
         */
    
        val sc = new SparkContext(conf) //创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息  
    
        /**
         * 第三步:根据具体的数据来源（HDFS，HBase，Local FS（本地文件系统） ，DB，S3（云上）等）通过SparkContext来创建RDD
         * RDD的创建基本有三种方式，根据外部的数据来源（例如HDFS），根据Scala集合，由其他的RDD操作产生
         * 数据会被RDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
         */
    
        //文件的路径，最小并行度（根据机器数量来决定）  
        //val lines:RDD[String]= sc.textFile("F://spark//spark-1.6.2-bin-hadoop2.6//README.md", 1)    //读取本地文件，并设置Partition = 1  
        val lines = sc.textFile("D://Java//spark//spark-1.6.1-bin-hadoop2.6//README.md", 1) //读取本地文件，并设置Partition = 1   //类型推导得出lines为RDD  
        /**
         * 第四步:对初始的RDD进行Transformation级别的处理，例如map，filter等高阶函数等的编程，来进行具体的数据计算
         *    4.1:将每一行的字符串拆分成单个的单词
         *    4.2:在单词拆分的基础上对每个单词的实例计数为1，也就是word =>(word,1)
         *    4.3:在每个单词实例计数为1基础之上统计每个单词在文件出现的总次数
         */
    
        //对每一行的字符串进行单词的拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合  
        val words = lines.flatMap { line => line.split(" ") } //words同样是RDD类型    
        val pairs = words.map { word => (word, 1) }
        val wordCounts = pairs.reduceByKey(_ + _) //对相同的key，进行value的累加（包括Local和Reducer级别同时Reduce）  
    
        wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))
    
        sc.stop() //注意一定要将SparkContext的对象停止，因为SparkContext运行时会创建很多的对象  
    
        /*这个程序运行之后一定会有一个错误，因为 没有hadoop环境，这个不是程序错误，也不影响任何功能*/
    
      }
    }

运行结果：

![20160823104538974][]

有可能会出现如下错误：

![20160823103941065][]

提示堆内存不够。右键-》run as ->run configurations 设置程序运行内存

![20160823103946331][]

2、远程运行实例

package com.lin.wordcount
    
      import org.apache.spark.SparkConf
     import org.apache.spark.SparkContext
     import org.apache.spark.SparkContext._
    
     /**
      * 在安装有Spark的Linux机器上跑
      * 环境：CentOS
      * scala版本：2.11.8
      * Spark版本：1.6.1
      * hadoop版本：hadoop2.6.4
      */
    object WordCount {
       def main(args: Array[String]) {
        if (args.length < 1) {
           System.err.println("Usage: <file>")
           System.exit(1)
         }
     
         val conf = new SparkConf()
         val sc = new SparkContext(conf)
         val line = sc.textFile(args(0))
     
         line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).collect().foreach(println)
         
         val result = line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)
        result.collect().foreach(println)
        result.saveAsSequenceFile(args(1)) 
     
         sc.stop()
       }
    }

看看pom文件：

<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>com.lin</groupId>
    	<artifactId>Spark-Demo</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<name>${project.artifactId}</name>
    	<description>My wonderfull scala app</description>
    	<inceptionYear>2015</inceptionYear>
    	<licenses>
    		<license>
    			<name>My License</name>
    			<url>http://....</url>
    			<distribution>repo</distribution>
    		</license>
    	</licenses>
    
    	<properties>
    		<maven.compiler.source>1.6</maven.compiler.source>
    		<maven.compiler.target>1.6</maven.compiler.target>
    		<encoding>UTF-8</encoding>
    		<scala.version>2.11.8</scala.version>
    		<scala.compat.version>2.11</scala.compat.version>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.scala-lang</groupId>
    			<artifactId>scala-library</artifactId>
    			<version>${scala.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-core_2.11</artifactId>
    			<version>1.6.1</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>4.11</version>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.specs2</groupId>
    			<artifactId>specs2-core_${scala.compat.version}</artifactId>
    			<version>2.4.16</version>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.scalatest</groupId>
    			<artifactId>scalatest_${scala.compat.version}</artifactId>
    			<version>2.2.4</version>
    			<scope>test</scope>
    		</dependency>
    	</dependencies>
    
    	<build>
    		<sourceDirectory>src/main/scala</sourceDirectory>
    		<testSourceDirectory>src/test/scala</testSourceDirectory>
    		<plugins>
    			<plugin>
    				
    				<groupId>net.alchim31.maven</groupId>
    				<artifactId>scala-maven-plugin</artifactId>
    				<version>3.2.0</version>
    				<executions>
    					<execution>
    						<goals>
    							<goal>compile</goal>
    							<goal>testCompile</goal>
    						</goals>
    						<configuration>
    							<args>
    								
    								<arg>-dependencyfile</arg>
    								<arg>${project.build.directory}/.scala_dependencies</arg>
    							</args>
    						</configuration>
    					</execution>
    				</executions>
    			</plugin>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-surefire-plugin</artifactId>
    				<version>2.18.1</version>
    				<configuration>
    					<useFile>false</useFile>
    					<disableXmlReport>true</disableXmlReport>
    					<includes>
    						<include>**/*Test.*</include>
    						<include>**/*Suite.*</include>
    					</includes>
    				</configuration>
    			</plugin>
    			<plugin>
    				<artifactId>maven-assembly-plugin</artifactId>
    				<version>2.6</version>
    				<configuration>
    					<descriptorRefs>
    						<descriptorRef>jar-with-dependencies</descriptorRef>
    					</descriptorRefs>
    				</configuration>
    			</plugin>
    		</plugins>
    	</build>
    	<profiles>
    		<profile>
    			<id>jdk-1.7</id>
    			<activation>
    				<activeByDefault>true</activeByDefault>
    				<jdk>1.7</jdk>
    			</activation>
    			<properties>
    				<maven.compiler.source>1.7</maven.compiler.source>
    				<maven.compiler.target>1.7</maven.compiler.target>
    				<maven.compiler.compilerVersion>1.7</maven.compiler.compilerVersion>
    			</properties>
    		</profile>
    	</profiles>
    
    </project>

项目目录结构：

![20160823104650209][]

运行;clean package打包：

![20160823104740764][]

将整个jar包上传到linux目录

![20160823104853585][]

运行脚本：

./bin/spark-submit --master  spark://hmaster:7077 --class com.lin.wordcount.WordCount --executor-memory 1g /app/hadoop/spark-1.6.1/examples/Spark-Demo-0.0.1-SNAPSHOT.jar /input/LICENSE.txt /output

输出结果：

![20160823104940064][]

同时将结果保存到hdfs中：

![20160823105001237][]

[20160823104538974]: /images/20210811/77b6b1831f39497d978081c04e2f2589.png
[20160823103941065]: /images/20210811/76b509ba6e854737b57ee941f2eb5f27.png
[20160823103946331]: /images/20210811/a96205f04add451bb7f8256f6021d767.png
[20160823104650209]: /images/20210811/aac9c74fcf30469f9449a3aff04b4120.png
[20160823104740764]: /images/20210811/ab6d24ddcb204373a1b2fdfe68c8d0f6.png
[20160823104853585]: /images/20210811/c3f367265ad94a2186807ff147df9066.png
[20160823104940064]: /images/20210811/c5eae471b7a54ed891e92d8c77ffd9c6.png
[20160823105001237]: /images/20210811/86ee980904c743d89ef42becb1595a72.png