Spark-了解以及快速入门

梦里梦外; 2022-09-02 04:20 168阅读 0赞

# Spark概述 #

## 什么是Spark ##

Spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

## Spark和Hadoop的区别 ##

Spark 和Hadoop 的区别：

**Hadoop**

Hadoop 是由 java 语言编写的，在分布式服务器集群上存储海量数据并运行分布式分析应用的开源框架，专用于数据批处理的框架，有存储也有计算，但是核心是计算且是离线计算。

作为 Hadoop 分布式文件系统，HDFS 处于 Hadoop 生态圈的最下层，存储着所有的数据， 支持着 Hadoop 的所有服务。它的理论基础源于 Google 的TheGoogleFileSystem 这篇论文，它是GFS 的开源实现。

MapReduce 是一种编程模型，Hadoop 根据 Google 的 MapReduce 论文将其实现， 作为 Hadoop 的分布式计算模型，是 Hadoop 的核心。基于这个框架，分布式并行程序的编写变得异常简单。综合了 HDFS 的分布式存储和 MapReduce 的分布式计算，Hadoop 在处理海量数据时，性能横向扩展变得非常容易。

HBase 是对 Google 的 Bigtable 的开源实现，但又和 Bigtable 存在许多不同之处。HBase 是一个基于HDFS 的分布式数据库，擅长实时地随机读/写超大规模数据集。它也是 Hadoop 非常重要的组件。

**Spark**

Spark 是一种由 Scala 语言开发的快速、通用、可扩展的大数据分析引擎，是包含流处理的批处理框架，侧重于通过内存来进行计算。

Spark Core 中提供了 Spark 最基础与最核心的功能

Spark SQL 是Spark 用来操作结构化数据的组件。通过 Spark SQL，用户可以使用SQL 或者 Apache Hive 版本的 SQL 方言（HQL）来查询数据。

Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的API。

Spark能够独立的进行集群部署，也可以取代MR与Hadoop进行集成部署。

Spark 出现的时间相对较晚，并且主要功能主要是用于数据计算， 所以其实 Spark 一直被认为是Hadoop 框架的升级版。

那么Hadoop 的 MR 框架和Spark 框架都是数据处理框架，我们在使用时如何选择呢？

1、Hadoop MapReduce 由于其设计初衷并不是为了满足循环迭代式数据流处理，因此在多并行运行的数据可复用场景（如：机器学习、图挖掘算法、交互式数据挖掘算法）中存在诸多计算效率等问题。所以 Spark 应运而生，Spark 就是在传统的MapReduce 计算框架的基础上，利用其计算过程的优化，从而大大加快了数据分析、挖掘的运行和读写速度，并将计算单元缩小到更适合并行计算和重复使用的RDD 计算模型。

> Hadoop的设计初衷是一次性数据计算，框架在处理数据的时候，会从存储设备中读取数据，并进行逻辑操作，然后讲处理的结果重新存储到介质中。

2、机器学习中 ALS、凸优化梯度下降等。这些都需要基于数据集或者数据集的衍生数据反复查询反复操作。MR 这种模式不太合适，即使多 MR 串行处理，性能和时间也是一个问题。数据的共享依赖于磁盘。另外一种是交互式数据挖掘，MR 显然不擅长。而Spark 所基于的 scala 语言恰恰擅长函数的处理。

3、Spark 是一个分布式数据快速分析项目。它的核心技术是弹性分布式数据集（Resilient Distributed Datasets），提供了比MapReduce 丰富的模型，可以快速在内存中对数据集进行多次迭代，来支持复杂的数据挖掘算法和图形计算算法。

4、**Spark 和Hadoop 的根本差异是多个作业之间的数据通信问题 : Spark 多个作业之间数据通信是基于内存，而 Hadoop 是基于磁盘**。

5、Spark Task 的启动时间快。Spark 采用 fork 线程的方式，而 Hadoop 采用创建新的进程的方式。

6、Spark 只有在 shuffle 的时候将数据写入磁盘，而 Hadoop 中多个 MR 作业之间的数据交互都要依赖于磁盘交互

7、Spark 的缓存机制比HDFS 的缓存机制高效。

经过上面的比较，我们可以看出在绝大多数的数据计算场景中，Spark 确实会比 MapReduce 更有优势。但是Spark 是基于内存的，所以在实际的生产环境中，由于内存的限制，可能会由于内存资源不够导致 Job 执行失败，此时，MapReduce 其实是一个更好的选择，所以 Spark 并不能完全替代 MR。

## Spark核心模块 ##

![在这里插入图片描述][b2fb12756cd841f9a02b23c83ff46829.png]

**Spark Core**

Spark Core 中提供了 Spark 最基础与最核心的功能，Spark 其他的功能如：Spark SQL， Spark Streaming，GraphX, MLlib 都是在 Spark Core 的基础上进行扩展的

**Spark SQL**

Spark SQL 是Spark 用来操作结构化数据的组件。通过 Spark SQL，用户可以使用 SQL或者Apache Hive 版本的 SQL 方言（HQL）来查询数据。

**Spark Streaming**

Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的API。

**Spark MLlib**

MLlib 是 Spark 提供的一个机器学习算法库。MLlib 不仅提供了模型评估、数据导入等额外的功能，还提供了一些更底层的机器学习原语。这个是进阶的功能，本次不涉及

**Spark GraphX**

GraphX 是 Spark 面向图计算提供的框架与算法库。这个是进阶的功能，本次不涉及

# Spark快速上手 #

## 环境准备 ##

IDEA中先保证安装了scala的插件，创建好maven工程，然后导入scala-sdk，导入scala-sdk的话通过File->ProjectStructure->Global Libraries中添加scala-sdk，版本号选择2.12.11。

然后为创建好的maven工程设置环境，在工程名右键->Add Frameworks Support，勾上Scala，然后确认。

验证scala环境是否运用成功，可以创建一个简单的scala程序

object Test { 
    
      def main(args: Array[String]): Unit = { 
        print("hello")
      }
    }

运行之后如果控制台输出了hello说明运用成功了。

## WordCount简单案例 ##

数据分析和操作的过程：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

在pom文件中加入依赖如下：

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
    </dependencies>

代码部分如下：

object WordCount { 
    
        def main(args: Array[String]): Unit = { 
    
            // TODO: 建立和Spark框架的连接
            // 本地模式
            val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparkConf)
    
            // TODO: 执行业务操作
            // 1、读取文件，获取一行一行的数据
            // hello word
            val lines = sc.textFile("datas/spark-core/wordCount")
    
            // 2、讲一行的数据进行拆分，形成一个一个的单词（分词）
            // 扁平化：将整体拆分成个体的操作
            // hello, word, hello, word
            val words = lines.flatMap(_.split(" "))
    
            // 3、将数据根据单词进行分组，便于统计
            // (hello, hello), (word, word)
            val wordGroup = words.groupBy(word => word)
    
            // 4、对分组后的数据进行转换
            // (hello, 2), (word, 2)
            val wordToCount = wordGroup.map { 
                case (word, list) => { 
                    (word, list.size)
                }
            }
    
            // 5、讲转换结果采集到控制台打印出来
            val array = wordToCount.collect()
            array.foreach(println)
    
            // TODO: 关闭连接
            sc.stop()
        }
    }

文件内容如下：

# 1.txt
    hello caocao
    hello liubei
    
    # 2.txt
    hello caocao
    hello liubei

运行结果如下：

![在这里插入图片描述][bbc7d994a39449a9be6943c9b04fc5d6.png]

进行思路的修改  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]

object WordCount2 { 
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparkConf)
    
            val lines = sc.textFile("datas/spark-core/wordCount")
    
            val words = lines.flatMap(_.split(" "))
    
            val wordToOne = words.map(word => (word, 1))
    
            val wordGroup = wordToOne.groupBy(word => word._1)
    
            val wordCount =  wordGroup.map { 
                case (word, list) => { 
                    val tuple = list.reduce((a, b) => { 
                        (word, a._2 + b._2)
                    })
                    tuple
                }
            }
    
            wordCount.foreach(println)
    
            sc.stop()
        }
    }

使用spark自带的reduceByKey方法

object WordCount3 { 
    
        def main(args: Array[String]): Unit = { 
    
            val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
            val sc = new SparkContext(sparkConf)
    
            val lines = sc.textFile("datas/spark-core/wordCount")
    
            val words = lines.flatMap(_.split(" "))
    
            val wordToOne = words.map(word => (word, 1))
    
            // Spark框架提供了更多的功能，可以将分组和聚合使用一个方法实现
            // reduceByKey: 相同的key的数据，可以对value进行reduce聚合
            // wordToOne.reduceByKey((x, y) => { x + y })
            val wordCount = wordToOne.reduceByKey(_ + _)
    
            wordCount.foreach(println)
    
            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

[b2fb12756cd841f9a02b23c83ff46829.png]: /images/20220829/783415d434cc421cacdadcb460ce4e48.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20220829/c737df3376af49b0a217cc6d75781a93.png
[bbc7d994a39449a9be6943c9b04fc5d6.png]: /images/20220829/2c379e773b7e44488f0b1e94b54a7c59.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70 1]: /images/20220829/02be809e6aea452592dad286bef53b97.png