Spark内存并行计算框架

• spark核心概念

• spark集群架构

• spark集群安装部署

• spark-shell的使用

• 通过IDEA开发spark程序

1. Spark是什么

Apache Spark™ is a unified analytics engine for large-scale data processingspark是针对于大规模数据处理的统一分析引擎

spark是在Hadoop基础上的改进，是UC Berkeley AMP lab所开源的类Hadoop MapReduce的通用的并行计算框架，Spark基于map reduce算法实现的分布式计算，拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是Job中间输出和结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的map reduce的算法。 ​ spark是基于内存计算框架，计算速度非常之快

输入 计算 输出

分发 计算 收集

第一步 保存中间结果 第二步

2. Spark

2.1 速度快

相对于hadoop来说，有使用hadoop MR经验会有体会

• 运行速度提高100倍

  Apache Spark使用最先进的DAG调度程序，查询优化程序和物理执行引擎，实现批量和流式数据的高性能。

• spark比mapreduce快的2个主要原因

  基于内存

  mapreduce任务后期再计算的时候，每一个job的输出结果会落地到磁盘，后续有其他的job需要依赖于前面job的输出结果，这个时候就需要进行大量的磁盘io操作。性能就比较低。 ​ spark任务后期再计算的时候，job的输出结果可以保存在内存中，后续有其他的job需要依赖于前面job的输出结果，这个时候就直接从内存中获取得到，避免了磁盘io操作，性能比较高

  进程与线程

  mapreduce任务以进程的方式运行在yarn集群中，比如程序中有100个MapTask，一个task就需要一个进程，这些task要运行就需要开启100个进程。 ​ spark任务以线程的方式运行在进程中，比如程序中有100个MapTask，后期一个task就对应一个线程，这里就不在是进程，这些task需要运行，这里可以极端一点：只需要开启1个进程，在这个进程中启动100个线程就可以了。进程中可以启动很多个线程，而开启一个进程与开启一个线程需要的时间和调度代价是不一样。 开启一个进程需要的时间远远大于开启一个线程。

2.2 易用性

• 可以快速去编写spark程序通过 java/scala/python/R/SQL等不同语言

2.3 通用性

• spark框架不在是一个简单的框架，可以把spark理解成一个生态系统，它内部是包含了很多模块，基于不同的应用场景可以选择对应的模块去使用

  sparksql、api

  通过sql去开发spark程序做一些离线分析

  sparkStreaming

  主要是用来解决公司有实时计算的这种场景

  Mlib

  它封装了一些机器学习的算法库

  Graphx 图计算

2.4 兼容性

• spark程序就是一个计算逻辑程序，这个任务要运行就需要计算资源（内存、cpu、磁盘），哪里可以给当前这个任务提供计算资源，就可以把spark程序提交到哪里去运行

  standAlone

  少量的节点，比较纯粹的spark任务

  它是spark自带的集群模式，整个任务的资源分配由spark集群的老大Master负责

  yarn

  yarn有接口，可以对接其他的计算框架 mr flink spark java

  可以把spark程序提交到yarn中运行，整个任务的资源分配由yarn中的老大ResourceManager负责

  mesos

  云原生支持的比较快

  它也是apache开源的一个类似于yarn的资源调度平台。

3. Spark与Yarn

mesos standalone yarn cpu 磁盘 内存

spark提交任务到yarn

• Driver

  它会执行客户端写好的main方法，它会构建一个名叫SparkContext对象

• Task

  spark任务是以task线程的方式运行在worker节点对应的executor进程中

spark是调用了yarn的接口，将任务提交给yarn来执行计算，实际的计算逻辑就是我们driver中的代码

4. Spark集群安装部署

• 事先搭建好hadoop分布式环境

• 下载安装包

• 解压安装包到指定的安装目录/root

• 重命名解压目录

  mv spark-3.1.1-bin-hadoop3.2 spark

• 修改配置文件，进入到spark的安装目录下对应的conf文件夹

  vim spark-env.sh ( mv spark-env.sh.template spark-env.sh)

  vim slaves ( mv slaves.template salves)

  #指定spark集群的worker节点
  hadoop01
  hadoop02
  hadoop03

• 分发安装目录到其他机器

  xsync /root/spark /root/spark

• 8、修改spark环境变量(也可以不配置，这里主要是为了)

  vim /etc/profile

  export SPARK_HOME=/root/spark
  export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

当启动好spark集群之后，可以访问这样一个地址http://hadoop01:8080

5. Spark怎么使用

5.1 生产环境

bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
examples/jars/spark-examples_2.12-3.1.1.jar \
10
​
##参数说明
--class：指定包含main方法的主类
--master：指定spark集群master地址
--executor-memory：指定任务在运行的时候需要的每一个executor内存大小
--total-executor-cores： 指定任务在运行的时候需要总的cpu核数
​

5.2 临时调试环境

spark整合HDFS，在hadoop01节点vim /root/spark/conf/spark-env.sh

export HADOOP_CONF_DIR=/opt/bigdata/hadoop/etc/hadoop

分发到其他节点

xsync /root/spark/conf/spark-env.sh /root/spark/conf/spark-env.sh

执行spark-shell --master yarn

sc.textFile("/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
​
sc.textFile("hdfs://hadoop01:9000/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

5.3 开发环境

构建maven工程，添加pom依赖

 <dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.1.1</version>
    </dependency>
</dependencies>
​
 <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass></mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
 </build>

提交任务使用的脚本

spark-submit \
--class com.mejava.day02.WordCount \
--master yarn \
--executor-memory 1G \
scala_learn-1.0-SNAPSHOT.jar