Spark学习--2.Spark快速上手-CFANZ编程社区

在上一篇文章中已经讲解了MapReduce 框架的原理及基本使用，并了解了其底层数据处理的实现方式。接下来，就让咱们走进Spark的世界，了解一下它是完成数据处理的。

2.1 创建Maven项目

2.1.1 增加 Scala 插件 Spark 由 Scala 语言开发的，所以接下来的开发所使用的语言也为Scala，当前使用的Spark版本为3.0.0，默认采用的Scala编译版本为2.12，所以后续开发时。采用这个版本。保证IDEA开发工具中含有Scala开发插件

Spark学习--2.Spark快速上手_apache

2.1.2 增加依赖关系

修改Maven项目中的POM文件，增加Spark框架的依赖关系。基于Spark3.0版本，使用时请注意对应版本。

<dependencies> 
    <dependency> 
    <groupId>org.apache.spark</groupId> 
    <artifactId>spark-core_2.12</artifactId> 
    <version>3.0.0</version> 
    </dependency> 
	</dependencies> 
	<build> 
		<plugins> 
		<!-- 该插件用于将Scala代码编译成class文件 --> 
			<plugin> 
        <groupId>net.alchim31.maven</groupId> 
        <artifactId>scala-maven-plugin</artifactId> 
        <version>3.2.2</version> 
        <executions> 
           <execution> 
              <!-- 声明绑定到maven的compile阶段 --> 
              <goals> 
              <goal>testCompile</goal> 
              </goals> 
          </execution> 
   			</executions> 
      </plugin> 
      <plugin> 
            <groupId>org.apache.maven.plugins</groupId> 
            <artifactId>maven-assembly-plugin</artifactId> 
            <version>3.1.0</version> 
            <configuration> 
                <descriptorRefs> 
                    <descriptorRef>jar-with-dependencies</descriptorRef> 
                </descriptorRefs> 
            </configuration> 
            <executions> 
                <execution> 
                    <id>make-assembly</id> 
                    <phase>package</phase> 
                    <goals> 
                        <goal>single</goal> 
                    </goals> 
                </execution> 
            </executions> 
        </plugin> 
    </plugins> 
</build>

2.1.3 WordCount

为了能直观地感受Spark框架的效果，实现一个最常见的案例WordCount

// 创建Spark运行配置对象 
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount") 
 
// 创建Spark上下文环境对象（连接对象） 
val sc : SparkContext = new SparkContext(sparkConf) 
 
// 读取文件数据 
val fileRDD: RDD[String] = sc.textFile("input/word.txt") 
 
// 将文件中的数据进行分词 
val wordRDD: RDD[String] = fileRDD.flatMap( _.split(" ") ) 
 
// 转换数据结构 word => (word, 1) 
val word2OneRDD: RDD[(String, Int)] = wordRDD.map((_,1)) 
 
// 将转换结构后的数据按照相同的单词进行分组聚合 
val word2CountRDD: RDD[(String, Int)] = word2OneRDD.reduceByKey(_+_) 
 
// 将数据聚合结果采集到内存中 
val word2Count: Array[(String, Int)] = word2CountRDD.collect() 
 
// 打印结果 
word2Count.foreach(println) 
 
//关闭Spark连接 
sc.stop()

执行过程中，会产生大量的执行日志，如果为了能够更好的查看程序的执行结果，可以在项目的resources目录中创建log4j.properties文件，并添加日志配置信息：

log4j.rootCategory=ERROR, console 
log4j.appender.console=org.apache.log4j.ConsoleAppender 
log4j.appender.console.target=System.err 
log4j.appender.console.layout=org.apache.log4j.PatternLayout 
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd 
HH:mm:ss} %p %c{1}: %m%n 
 
# Set the default spark-shell log level to ERROR. When running the spark-shell, 
the 
# log level for this class is used to overwrite the root logger's log level, so 
that 
# the user can have different defaults for the shell and regular Spark apps. 
log4j.logger.org.apache.spark.repl.Main=ERROR 
 
# Settings to quiet third party logs that are too verbose 
log4j.logger.org.spark_project.jetty=ERROR 
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR 
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=ERROR 
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=ERROR 
log4j.logger.org.apache.parquet=ERROR 
log4j.logger.parquet=ERROR 
 
# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent 
UDFs in SparkSQL with Hive support 
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL 
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR

2.1.4 异常处理

如果本机操作系统是Windows，在程序中使用了Hadoop相关的东西，比如写入文件到 HDFS，则会遇到如下异常：

Spark学习--2.Spark快速上手_spark_02