第一课 大数据技术之Spark-环境和快速入门

文章目录

第一节 Spark 概述

1.1 Spark介绍

  1. Spark 是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。
  2. 首先从时间节点上来看Hadoop和Spark
  3. Hadoop
    • 2006 年 1 月，Doug Cutting 加入Yahoo，领导Hadoop 的开发
    • 2008 年 1 月，Hadoop 成为 Apache 顶级项目
    • 2011 年 1.0 正式发布
    • 2012 年 3 月稳定版发布
    • 2013 年 10 月发布 2.X (Yarn)版本
  4. Spark
    • 2009 年，Spark 诞生于伯克利大学的AMPLab 实验室
    • 2010 年，伯克利大学正式开源了 Spark 项目
    • 2013 年 6 月，Spark 成为了 Apache 基金会下的项目
    • 2014 年 2 月，Spark 以飞快的速度成为了 Apache 的顶级项目
    • 2015 年至今，Spark 变得愈发火爆，大量的国内公司开始重点部署或者使用 Spark
  5. 然后我们再从功能上来看Hadoop和Spark
  6. 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 非常重要的组件。
  7. Spark
    • Spark 是一种由 Scala 语言开发的快速、通用、可扩展的大数据分析引擎
    • Spark Core 中提供了 Spark 最基础与最核心的功能
    • Spark SQL 是Spark 用来操作结构化数据的组件。通过 Spark SQL，用户可以使用SQL 或者 Apache Hive 版本的 SQL 方言（HQL）来查询数据。
    • Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的API。
  8. 由上面的信息可以获知，Spark 出现的时间相对较晚，并且主要功能主要是用于数据计算， 所以其实 Spark 一直被认为是Hadoop 框架的升级版。

1.2 Spark和Hadoop选择

  1. Hadoop 的 MR 框架和Spark 框架都是数据处理框架，那么我们在使用时如何选择呢？
  2. Hadoop MapReduce 由于其设计初衷并不是为了满足循环迭代式数据流处理，因此在多并行运行的数据可复用场景（如：机器学习、图挖掘算法、交互式数据挖掘算法）中存在诸多计算效率等问题。所以 Spark 应运而生，Spark 就是在传统的MapReduce 计算框架的基础上，利用其计算过程的优化，从而大大加快了数据分析、挖掘的运行和读写速度，并将计算单元缩小到更适合并行计算和重复使用的RDD 计算模型。Hadoop的初衷是为了一次性的数据计算。
    在这里插入图片描述
  3. 机器学习中 ALS、凸优化梯度下降等。这些都需要基于数据集或者数据集的衍生数据反复查询反复操作。MR 这种模式不太合适，即使多 MR 串行处理，性能和时间也是一个问题。数据的共享依赖于磁盘。另外一种是交互式数据挖掘，MR 显然不擅长。而Spark 所基于的 scala 语言恰恰擅长函数的处理。
  4. Spark 是一个分布式数据快速分析项目。它的核心技术是弹性分布式数据集（Resilient Distributed Datasets），提供了比MapReduce 丰富的模型，可以快速在内存中对数据集进行多次迭代，来支持复杂的数据挖掘算法和图形计算算法。
  5. Spark 和Hadoop 的根本差异是多个作业之间的数据通信问题 : Spark 多个作业之间数据通信是基于内存，而 Hadoop 是基于磁盘。
  6. Spark Task 的启动时间快。Spark 采用 fork 线程的方式，而 Hadoop 采用创建新的进程的方式。
  7. Spark 只有在 shuffle 的时候将数据写入磁盘，而 Hadoop 中多个 MR 作业之间的数据交互都要依赖于磁盘交互
  8. Spark 的缓存机制比HDFS 的缓存机制高效。
  9. 经过上面的比较，我们可以看出在绝大多数的数据计算场景中，Spark 确实会比 MapReduce 更有优势。但是Spark 是基于内存的，所以在实际的生产环境中，由于内存的限制，可能会由于内存资源不够导致 Job 执行失败，此时，MapReduce 其实是一个更好的选择，所以 Spark 并不能完全替代 MR。

1.3 核心模块介绍

在这里插入图片描述

  1. Spark Core。Spark Core 中提供了 Spark 最基础与最核心的功能，Spark 其他的功能如：Spark SQL， Spark Streaming，GraphX, MLlib 都是在 Spark Core 的基础上进行扩展的
  2. Spark SQL。Spark SQL 是Spark 用来操作结构化数据的组件。通过 Spark SQL，用户可以使用 SQL或者Apache Hive 版本的 SQL 方言（HQL）来查询数据。
  3. Spark Streaming。Spark Streaming 是 Spark 平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的API。
  4. Spark MLlib。MLlib 是 Spark 提供的一个机器学习算法库。MLlib 不仅提供了模型评估、数据导入等额外的功能，还提供了一些更底层的机器学习原语。
  5. Spark GraphX。GraphX 是 Spark 面向图计算提供的框架与算法库。

第二节 环境搭建和快速上手

2.1 增加 Scala 插件

  1. Spark 由 Scala 语言开发的，开发需要引入Scala插件，咱们开发所使用的语言也为 Scala,当前使用的Spark 版本为 3.0.0，默认采用的 Scala 编译版本为 2.12，所以后续开发时。我们依然采用这个版本。开发前请保证 IDEA 开发工具中含有 Scala 开发插件。
  2. https://www.scala-lang.org/download/2.12.11.html 下载msi，安装Idea插件Scala并重启。配置全局类库File->project Structure
    在这里插入图片描述
  3. 创建一个Scala测试类。com.atguigu.bigdata.spark.core.Test
package com.atguigu.bigdata.spark.core

object Test {

    def main(args: Array[String]): Unit = {

        println("Hello Spark")

    }
}
  1. 运行测试下，看下环境是否正常。

2.2 快速上手wordcount

  1. 引入pom.
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
  1. 创建数据datas下两个文件。
# 1.txt
hello spark
hello scala

# 2.txt
hello spark
hello scala
  1. 写下代码com.atguigu.bigdata.spark.core.wc.Spark01_WordCount
package com.atguigu.bigdata.spark.core.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 sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        // TODO 执行业务操作

        // 1. 读取文件，获取一行一行的数据
        //    hello world
        val lines: RDD[String] = sc.textFile("datas")

        // 2. 将一行数据进行拆分，形成一个一个的单词（分词）
        //    扁平化：将整体拆分成个体的操作
        //   "hello world" => hello, world, hello, world
        val words: RDD[String] = lines.flatMap(_.split(" "))

        // 3. 将数据根据单词进行分组，便于统计
        //    (hello, hello, hello), (world, world)
        val wordGroup: RDD[(String, Iterable[String])] = words.groupBy(word=>word)

        // 4. 对分组后的数据进行转换
        //    (hello, hello, hello), (world, world)
        //    (hello, 3), (world, 2)
        val wordToCount = wordGroup.map {
            case ( word, list ) => {
                (word, list.size)
            }
        }

        // 5. 将转换结果采集到控制台打印出来
        val array: Array[(String, Int)] = wordToCount.collect()
        array.foreach(println)

        // TODO 关闭连接
        sc.stop()
    }
}

2.3 wordcount优化用聚合的逻辑

package com.atguigu.bigdata.spark.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark02_WordCount1 {

    def main(args: Array[String]): Unit = {

        // Application
        // Spark框架
        // TODO 建立和Spark框架的连接
        // JDBC : Connection
        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        // TODO 执行业务操作

        // 1. 读取文件，获取一行一行的数据
        //    hello world
        val lines: RDD[String] = sc.textFile("datas")

        // 2. 将一行数据进行拆分，形成一个一个的单词（分词）
        //    扁平化：将整体拆分成个体的操作
        //   "hello world" => hello, world, hello, world
        val words: RDD[String] = lines.flatMap(_.split(" "))

        // 3. 将单词进行结构的转换,方便统计
        // word => (word, 1)
        val wordToOne = words.map(word=>(word,1))

        // 4. 将转换后的数据进行分组聚合
        // spark的方法reduceByKey: 相同key的value进行聚合操作
        // (word, 1) => (word, sum)
        val wordToSum: RDD[(String, Int)] = wordToOne.reduceByKey(_+_)

        // 5. 将转换结果采集到控制台打印出来
        val array: Array[(String, Int)] = wordToSum.collect()
        array.foreach(println)

        // TODO 关闭连接
        sc.stop()

    }
}

2.4 wordcount常用方法

package com.atguigu.bigdata.spark.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Spark03_WordCount {
    def main(args: Array[String]): Unit = {

        val sparConf = new SparkConf().setMaster("local").setAppName("WordCount")
        val sc = new SparkContext(sparConf)

        wordcount91011(sc)

        sc.stop()

    }

    // groupBy
    def wordcount1(sc : SparkContext): Unit = {

        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val group: RDD[(String, Iterable[String])] = words.groupBy(word=>word)
        val wordCount: RDD[(String, Int)] = group.mapValues(iter=>iter.size)
    }

    // groupByKey
    def wordcount2(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val group: RDD[(String, Iterable[Int])] = wordOne.groupByKey()
        val wordCount: RDD[(String, Int)] = group.mapValues(iter=>iter.size)
    }

    // reduceByKey
    def wordcount3(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val wordCount: RDD[(String, Int)] = wordOne.reduceByKey(_+_)
    }

    // aggregateByKey
    def wordcount4(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val wordCount: RDD[(String, Int)] = wordOne.aggregateByKey(0)(_+_, _+_)
    }

    // foldByKey
    def wordcount5(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val wordCount: RDD[(String, Int)] = wordOne.foldByKey(0)(_+_)
    }

    // combineByKey
    def wordcount6(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val wordCount: RDD[(String, Int)] = wordOne.combineByKey(
            v=>v,
            (x:Int, y) => x + y,
            (x:Int, y:Int) => x + y
        )
    }

    // countByKey
    def wordcount7(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordOne = words.map((_,1))
        val wordCount: collection.Map[String, Long] = wordOne.countByKey()
    }

    // countByValue
    def wordcount8(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))
        val wordCount: collection.Map[String, Long] = words.countByValue()
    }

    // reduce, aggregate, fold
    def wordcount91011(sc : SparkContext): Unit = {
        val rdd = sc.makeRDD(List("Hello Scala", "Hello Spark"))
        val words = rdd.flatMap(_.split(" "))

        // 【（word, count）,(word, count)】
        // word => Map[(word,1)]
        val mapWord = words.map(
            word => {
                mutable.Map[String, Long]((word,1))
            }
        )

       val wordCount = mapWord.reduce(
            (map1, map2) => {
                map2.foreach{
                    case (word, count) => {
                        val newCount = map1.getOrElse(word, 0L) + count
                        map1.update(word, newCount)
                    }
                }
                map1
            }
        )

        println(wordCount)
    }

}

第三节 执行日志和常见异常

3.1 执行日志

  1. 执行过程中，会产生大量的执行日志，如果为了能够更好的查看程序的执行结果。可以在项目的 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

3.2 常见异常

  1. 如果本机操作系统是 Windows，在程序中使用了 Hadoop 相关的东西，比如写入文件到HDFS，则会遇到如下异常：
    在这里插入图片描述
  2. 出现这个问题的原因，并不是程序的错误，而是windows 系统用到了 hadoop 相关的服务，解决办法是通过配置关联到 windows 的系统依赖就可以了。在 IDEA 中配置Run Configuration，添加HADOOP_HOME变量
# spark
Logo
技术共进，成长同行——讯飞AI开发者社区

技术共进，成长同行——讯飞AI开发者社区

更多推荐

cover

YOLOv8【主干网络篇·第4节】LSKNet大核卷积遥感检测专用网络！

avatar 讯飞AI开发者社区

**发散创新：开源商业中的编程语言探索——以XX语言为例**在开源商业领域，编程语言的选择至关重

本文介绍了XX语言的特点及其在开源电商管理平台项目中的应用。通过实际案例，展示了XX语言在开源商业领域的优势。随着开源商业的不断发展，我们将继续探索更多编程语言在开源项目中的应用，为开发者提供更多有价值的参考。本文将带领大家了解XX语言的特点及其在开源项目中的应用。本文将深入探讨如何在开源项目中合理运用编程语言，并结合XX语言的特点，为大家展示一个充满创新的项目实例。XX语言是一种高效、易学、功能

avatar 讯飞AI开发者社区

人工智能与大数据在2024年医疗健康领域的十大革命性应用场景

而匿名的患者大数据分享平台，则让患有相同疾病的患者能够交换“真实世界”的治疗经验和效果数据，形成宝贵的互助网络，为个体治疗决策提供了强大的社群支持，推动了以患者为中心的医疗模式发展。更重要的是，AI正在辅助设计更安全、高效的CRISPR基因编辑工具，预测编辑脱靶效应，加速基因治疗从基础研究走向临床应用的进程，为根治遗传性疾病开启了新的大门。同时，在供应链管理中，AI可预测药品和医疗物资的需求，实现

avatar 讯飞AI开发者社区