9.2.1、Spark Core__spark概述，spark与mapreduce的对比，核心模块，运行环境，端口号，spark的结构组件，wordcount案例，运行机制client，cluster

2024-02-08 13:51:16

1、spqrk概述

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

2、与hadoop的对比

（1）hadoop是在2006年出现的，java编写
（2）spark在2009年诞生于伯克利大学实验室，基于内存的大数据计算引擎，数据还是存储与hadfs上
（3）Spark 和Hadoop 的根本差异是多个作业之间的数据通信问题 : Spark 多个作业之间数据通信是基于内存，而 Hadoop 是基于磁盘。

3、核心模块

Spark Core：Spark 最基础与最核心的功能，Spark 其他的功能都是在此基础上进行扩展的

Spark SQL：是 Spark 用来操作结构化数据的组件，可以操作sql语句

Spark Streaming：是 Spark 平台上针对实时数据进行流式计算的组件
Spark MLlib：MLlib 是 Spark 提供的一个机器学习算法库

Spark GraphX：GraphX 是 Spark 面向图计算提供的框架与算法库

4、运行环境

（1）local本地模式
（2）standalone模式：Spark 的 Standalone 模式体现了经典的 master-slave
模式集群规划：master，worker主从节点
（3）mesos模式：和hadoop差不多
（4）yarn：使用hadoop的yarn做任务调度集群运行

5、端口号

Spark 查看当前 Spark-shell 运行任务情况端口号：4040（计算） Spark Master 内部通信服务端口号：7077

Standalone 模式下，Spark Master Web 端口号：8080（资源）

Spark 历史服务器端口号：18080

Hadoop YARN 任务运行情况查看端口号：8088

6、spark架构组件

（1）计算的组件
Driver：计算的主节点(分区，形成独立的task)
Executor：一个JVM进程，计算的从节点
（2）调度的组件（standalone）
Master：调度的主节点（相当于ResourceManager）
Worker：调度的从节点（相当于nodeManager）
（3）ApplicationMaster：连接计算和调度的组件

7、wordcount案例

（1）导包，同时需要导入scala需要的包

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.4.5</version>
</dependency>

（2）配置环境变量，不配置可能会报一个错误

新建文件夹HAdoop/bin，里面放入winutils.exe文件，配置环境变量新建HADOOP_HOME=HAdoop路径(新建的)

（3）示例

object Demo01wordcount {
  def main(args: Array[String]): Unit = {
    //程序的配置文件对象
    val conf: SparkConf = new SparkConf()
    ///程序的名称
    conf.setAppName("Demo01wordcount")
    //程序的运行模式
    conf.setMaster("local")
    //程序的上下文对象，程序的入口
    val sc: SparkContext = new SparkContext(conf)
    val lineRDD: RDD[String] = sc.textFile("sparkproject/data/words.txt",2)
    //filePath.foreach(println)
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))
    val wordgroupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)
    val wordCountRDD: RDD[String] = wordgroupRDD.map(wordgroup => {
      val word: String = wordgroup._1
      val words: Iterable[String] = wordgroup._2
      word + "," + words.size
    })
    //将结果写回文件，要是再集群上就填hdfs的路径
    wordCountRDD.saveAsTextFile("sparkproject/data/res")
  }
}

8、wordcount打包在集群运行

（1）修改集群的读取输入输出位置为HDFS上位置
（2）将setMaster去除（在集群运行，不是在本地运行）
（3）上传Jar包

yarn-client模式：

spark-submit --clazz 主类名 --master yarn-client Jar包
yarn-cluster模式：(提交就行了，退出还是会运行) spark-submit --class 主类名 --master
yarn-cluster Jar包

读取文件可以指定一个minPartitions参数
textFile(“sparkproject/data/words.txt”,2)//指定最小分区默认是一个block对应一个分区
但是里面有一个限制minPartition(并行度，2) //集群并行度可能默认不是1
上传至集群运行就生效了，一个文件最终也是2个分区（运行拆开）

本地运行:

conf.setMaster(“local”) //一个并行度 conf.setMaster(“local[2]”) //两个并行度
conf.setMaster(“local[*]”) //有多少用多少并行度
本地运行不同的是，一个文件就是一个分区，多个文件的话就是一个文件一个分区
还是minPartition(并行度，2) //本地的并行度可能默认是1

9、运行机制

yarn-client

yarn-cluster

码农公寓