一、Spark 概述

1、Spark 简介

Spark 是专为大规模数据处理而设计的，基于内存疾速通用, 可扩大的集群计算引擎，实现了高效的 DAG 执行引擎, 能够通过基于内存来高效解决数据流，运算速度相比于 MapReduce 失去了显著的进步。

2、运行构造

Driver

运行 Spark 的 Applicaion 中 main()函数，会创立 SparkContext，SparkContext 负责和 Cluster-Manager 进行通信，并负责申请资源、任务分配和监控等。

ClusterManager

负责申请和治理在 WorkerNode 上运行利用所需的资源，能够高效地在一个计算节点到数千个计算节点之间伸缩计算，目前包含 Spark 原生的 ClusterManager、ApacheMesos 和 HadoopYARN。

Executor

Application 运行在 WorkerNode 上的一个过程，作为工作节点负责运行 Task 工作，并且负责将数据存在内存或者磁盘上，每个 Application 都有各自独立的一批 Executor，工作间互相独立。

二、环境部署

1、Scala 环境

安装包治理

[root@hop01 opt]# tar -zxvf scala-2.12.2.tgz
[root@hop01 opt]# mv scala-2.12.2 scala2.12

配置变量

[root@hop01 opt]# vim /etc/profile

export SCALA_HOME=/opt/scala2.12
export PATH=$PATH:$SCALA_HOME/bin

[root@hop01 opt]# source /etc/profile

版本查看

[root@hop01 opt]# scala -version

Scala 环境须要部署在 Spark 运行的相干服务节点上。

2、Spark 根底环境

安装包治理

[root@hop01 opt]# tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz
[root@hop01 opt]# mv spark-2.1.1-bin-hadoop2.7 spark2.1

配置变量

[root@hop01 opt]# vim /etc/profile

export SPARK_HOME=/opt/spark2.1
export PATH=$PATH:$SPARK_HOME/bin

[root@hop01 opt]# source /etc/profile

版本查看

[root@hop01 opt]# spark-shell

3、Spark 集群配置

服务节点

[root@hop01 opt]# cd /opt/spark2.1/conf/
[root@hop01 conf]# cp slaves.template slaves
[root@hop01 conf]# vim slaves

hop01
hop02
hop03

环境配置

[root@hop01 conf]# cp spark-env.sh.template spark-env.sh
[root@hop01 conf]# vim spark-env.sh

export JAVA_HOME=/opt/jdk1.8
export SCALA_HOME=/opt/scala2.12
export SPARK_MASTER_IP=hop01
export SPARK_LOCAL_IP= 装置节点 IP
export SPARK_WORKER_MEMORY=1g
export HADOOP_CONF_DIR=/opt/hadoop2.7/etc/hadoop

留神 SPARK_LOCAL_IP 的配置。

4、Spark 启动

依赖 Hadoop 相干环境，所以要先启动。

启动：/opt/spark2.1/sbin/start-all.sh
进行：/opt/spark2.1/sbin/stop-all.sh

这里在主节点会启动两个过程：Master 和 Worker，其余节点只启动一个 Worker 过程。

5、拜访 Spark 集群

默认端口是：8080。

http://hop01:8080/

运行根底案例：

[root@hop01 spark2.1]# cd /opt/spark2.1/
[root@hop01 spark2.1]# bin/spark-submit --class org.apache.spark.examples.SparkPi --master local examples/jars/spark-examples_2.11-2.1.1.jar

运行后果：Pi is roughly 3.1455357276786384

三、开发案例

1、外围依赖

依赖 Spark2.1.1 版本：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.1.1</version>
</dependency>

引入 Scala 编译插件：

<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>
        </execution>
    </executions>
</plugin>

2、案例代码开发

读取指定地位的文件，并输入文件内容单词统计后果。

@RestController
public class WordWeb implements Serializable {@GetMapping("/word/web")
    public String getWeb (){
        // 1、创立 Spark 的配置对象
        SparkConf sparkConf = new SparkConf().setAppName("LocalCount")
                                             .setMaster("local[*]");

        // 2、创立 SparkContext 对象
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        sc.setLogLevel("WARN");

        // 3、读取测试文件
        JavaRDD lineRdd = sc.textFile("/var/spark/test/word.txt");

        // 4、行内容进行切分
        JavaRDD wordsRdd = lineRdd.flatMap(new FlatMapFunction() {
            @Override
            public Iterator call(Object obj) throws Exception {String value = String.valueOf(obj);
                String[] words = value.split(",");
                return Arrays.asList(words).iterator();}
        });

        // 5、切分的单词进行标注
        JavaPairRDD wordAndOneRdd = wordsRdd.mapToPair(new PairFunction() {
            @Override
            public Tuple2 call(Object obj) throws Exception {
                // 将单词进行标记：return new Tuple2(String.valueOf(obj), 1);
            }
        });

        // 6、统计单词呈现次数
        JavaPairRDD wordAndCountRdd = wordAndOneRdd.reduceByKey(new Function2() {
            @Override
            public Object call(Object obj1, Object obj2) throws Exception {return Integer.parseInt(obj1.toString()) + Integer.parseInt(obj2.toString());
            }
        });

        // 7、排序
        JavaPairRDD sortedRdd = wordAndCountRdd.sortByKey();
        List<Tuple2> finalResult = sortedRdd.collect();

        // 8、后果打印
        for (Tuple2 tuple2 : finalResult) {System.out.println(tuple2._1 + "===>" + tuple2._2);
        }

        // 9、保留统计后果
        sortedRdd.saveAsTextFile("/var/spark/output");
        sc.stop();
        return "success" ;
    }
}

打包执行后果：

查看文件输入：

[root@hop01 output]# vim /var/spark/output/part-00000

四、源代码地址

GitHub·地址
https://github.com/cicadasmile/big-data-parent
GitEE·地址
https://gitee.com/cicadasmile/big-data-parent

浏览标签

【Java 根底】【设计模式】【构造与算法】【Linux 零碎】【数据库】

【分布式架构】【微服务】【大数据组件】【SpringBoot 进阶】【Spring&Boot 根底】

【数据分析】【技术导图】【职场】

技术系列

OLAP 引擎：Druid 组件进行数据统计分析

OLAP 引擎：Presto 组件跨数据源剖析

OLAP 引擎：ClickHouse 高性能列式查问