在此之前,我已经用MapReduce 框架实现了WordCount案例,接下来,我开始学习数据处理的另外一个非常重要的方法:Spark。首先,使用WordCount案例实现Spark快速上手。
创建Maven项目
1. 增加Scala插件
使用的Spark 版本为 3.0.0,默认采用的Scala 编译版本为 2.12
操作步骤
2. 增加依赖关系
修改 Maven 项目中的POM 文件,增加Spark 框架的依赖关系(请注意对应版本)
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
3. 添加日志配置信息
执行过程中,会产生大量的执行日志,如果为了能够更好的查看程序的执行结果,可以在项
目的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
注意:两个红框的地方都要有log4j.properties 文件(只需要在resources下创建就可以了,target目录下是自动生成的,如果没有生成,直接复制一份粘贴过去)
WordCount案例
1. 代码示例
package wc
import org.apache.spark.rdd.RDD
import org.apache.spark.{
SparkConf, SparkContext}
object Spark01_WordCount {
def main(args: Array[String]): Unit = {
//application
//Spark框架:是一个环境
//TODO 建立和spark的连接
//JDBC:Connection
val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
val sc = new SparkContext(sparkConf)
//TODO 执行业务操作
//1.读取文件,获取一行一行的数据
var lines: RDD[String] = sc.textFile("datas")
//2.一行数据进行拆分,形成一个一个的单词(分词):扁平化
var words: RDD[String] = lines.flatMap(_.split(" "))
//3.将数据根据分词进行分组,便于统计
var wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word => word)
//4.对分组后的数据进行转换(统计)
val wordToCount = wordGroup.map{
case (word, list) => {
(word, list.size)
}
}
//5.将转换结果采集到控制台打印出来
var varry: Array[(String, Int)] = wordToCount.collect()
varry.foreach(println)
//TODO 关闭spark的连接
sc.stop()
}
}