Spark内存计算实战练习

实验预计耗时：45分钟

1. 课程背景

1.1 课程目的

Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地集合对象一样轻松地操作分布式数据集。

本教程通过Spark实现WordCount程序的实验任务，带领学员掌握Spark内存计算框架，掌握基本的编程语法。

1.2 课前知识准备

学习本课程前，学员需要掌握以下前置知识：

1、能力基础

• Linux基本操作：掌握Linux远程登录、文件与目录管理、vim 编辑器使用等。
• Hadoop基础：理解Hadoop基本组件的功能与原理。
• Scala语法基础：包括SQL基本语法、数据类型、常用内置函数等。

2、相关技术

• Spark：是基于内存计算的大数据分布式计算框架。Spark基于内存计算，提高了在大数据环境下数据处理的实时性，同时保证了高容错性和高可伸缩性，允许用户将Spark部署在大量廉价硬件之上，形成集群。
  • 提供分布式计算功能，将分布式存储的数据读入，同时将任务分发到各个节点进行计算；
  • 基于内存计算，将磁盘数据读入内存，将计算的中间结果保存在内存，这样可以很好的进行迭代运算；
  • 支持高容错；
  • 提供多计算范式也叫算子；
• Spark运行模式
  • 本地模式：此种模式下，只要将Spark包在本地解压即可使用，运行时Spark目录下的bin目录执行bin/spark-shell即可。
  • Standalone模式：即独立模式，自带完整的资源管理和任务调度等功能，可独立部署一个集群，无需依赖任何其他资源管理系统。
  • Spark On YARN模式：是一种最有前景的部署模式。但限于Yarn的自身发展，目前仅支持粗粒度的模式。
  • Spark On Mesos模式：Spark运行在Mesos上会比运行在YARN上更加灵活和自燃。在Spark on Mesos环境中用户可选择两种调度模式之一运行自己的应用程序，粗粒度模式和细粒度模式。
• Yarn-cluster模式应用程序启动步骤
  • ResourceManager接到请求后在集群中选择一个NodeManager分配Container，并在Container中启动ApplicationMaster进程；
  • 在ApplicationMaster进程中初始化sparkContext；
  • ApplicationMaster向ResourceManager申请到Container后，通知NodeManager在获得的Container中启动excutor进程；
  • SparkContext分配Task给excutor，excutor发送运行状态给ApplicationMaster。

• 如今Spark针对不同的应用场景构建了Spark生态体系，其体系架构如下，而SparkCore也是通常说的Spark：

• RDD(Resilient Distributed Datasets)：弹性分布式数据集，是分布式内存的一个抽象概念，RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录分区的集合，只能通过在其他RDD执行确定的转换操作（如map、join和group by）而创建，然而这些限制使得实现容错的开销很低。对开发者而言，RDD可以看作是Spark的一个对象，它本身运行于内存中，如读文件是一个RDD，对文件计算是一个RDD，结果集也是一个RDD ，不同的分片、 数据之间的依赖 、key-value类型的map数据都可以看做RDD。

2. 实验环境

2.1 实验操作环境

本课程需要以下实验操作环境：

1. 可以接入互联网的笔记本电脑或者台式机，本实验使用Windows系统
2. 实验环境：计算机本地（需具备Java开发环境）+腾讯云控制台

2.2 实验架构图

本实验将开发的Spark程序以jar包的形式提交给EMR三节点集群（Master节点和两个Core节点），其架构图如下：

2.3 实验的数据规划表

Scala下载

PuTTY下载

3. 实验流程

本实验涵盖Spark开发环境准备，Spark程序编写打包，程序上传集群，Spark程序运行验证以及Spark任务记录查看共计5个步骤。

首先实验环境准备，需要借助腾讯云弹性MapReduce服务帮助我们快速搭建一个三节点的Spark集群，注意主节点最低配置为4核8G。接下来借助IntelliJ IDEA工具使用Scala快速开发一个Spark WordCount程序。

程序开发完成后我们使用Maven来快速打包，并使用PuTTY将程序包上传至集；接着获取实验数据，使用Spark-submit运行程序并查看运行结果。

最后我们可以在YARN的Web UI上查看到任务执行记录，查看任务运行信息。

相信通过本实验，学员可以了解到Spark程序在EMR集群上开发和运行的全部流程。

4. 实验步骤

任务1 实验环境准备

【任务目标】

通过EMR集群的搭建练习，使学员可以熟练搭建EMR集群，快速构建实验所需大数据平台环境。并在本地配置IDEA，创建一个使用Scala的Maven项目。

【任务步骤】

1、EMR创建Hadoop集群

1.在腾讯云官网，找到弹性MapReduce首页，点击立即购买。

2.可用区与软件配置如下：

确认配置无误后，点击下一步：硬件配置。

注意：请勿选择EMR-3.0.0 tlinux，如产品版本无EMR-3.0.0版本，请更换个人主账号进行实验。

3.硬件配置如下：

启动高可用选项可以自定义选择，默认是选择的，如果取消需要手动取消选择。由于我们这里的实验环境仅仅是一个学习的实验环境所以这里我们将此选项取消，实际生产中要根据实际环境合理选择是否需要这个配置。

确认硬件配置信息无误后，点击下一步：基础配置。

4.基础配置如下：

确认信息无误后，点击购买，会自动跳转至集群页。图中的集群实例状态中显示集群创建中。

等待5min左右，集群构建成功，截图如下：

2、IDEA配置Scala插件

1.若要使用IDEA时支持Scala，需要在IDEA中安装Scala插件，首页右下角点击Configure->Plugins ；

备注：Plugins选项配置IDEA的相关插件信息。

2.在Marketplace下的搜索框输入scala，点击回车；找到Scala插件后，点击Install。

3.您也可以离线下载相应版本的scala插件；

IDEA官网下载网址如下：https://plugins.jetbrains.com/plugin/1347-scala

在网页内选择对应自己IDEA版本的插件进行下载。

4.配置Scala插件，选择离线安装包后重启Idea重启后再次进入插件安装页进行安装即可。

备注：这里Scala插件的安装我们选择Install Plugin from Disk方式进行安装

3、创建Scala项目

1.IEDA创建Maven项目TestSpark。

2.项目中增加Scala支持；

右键点击项目名称，选择Add Framework Support...；

在Add Framework Support的左侧列表中选择Scala。

3.在main目录下创建目录scala，修改scala目录为source目录；

右键点击创建的scala目录，选择Mark Directory as中的Sources Root。

4.在scala目录下创建包com.test.bigdata，在包内创建一个Scala Class；

注意：这里选择Object类型。

5.编写以下测试代码并运行；

package com.test.bigdata

object TestScala {
  def main(args: Array[String]): Unit = {
    println("Hello Spark !!! ")
  }
}

编写代码无误后，右键点击Run 'TestScala'，正常显示结果如下：

测试代码可以正常运行，接下来就在这个项目的基础上进行实验。

任务2 Spark程序WordCount练习

【任务目标】

通过Spark实现WordCount的练习，掌握Spark的程序编写，任务打包提交，任务运行与结果查看的全流程。

【任务步骤】

1、程序编写与打包

1.在pom.xml中添加依赖和打包插件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.test</groupId>
    <artifactId>TestSpark</artifactId>
    <version>1.0-SNAPSHOT</version>
 
    <repositories>
        <repository>
            <id>scala-tools.org</id>
            <name>Scala-Tools Maven2 Repository</name>
            <url>http://scala-tools.org/repo-releases</url>
        </repository>
    </repositories>

    <pluginRepositories>
        <pluginRepository>
            <id>scala-tools.org</id>
            <name>Scala-Tools Maven2 Repository</name>
            <url>http://scala-tools.org/repo-releases</url>
        </pluginRepository>
    </pluginRepositories>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.8</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.4.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-yarn_2.11</artifactId>
            <version>2.4.3</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.11</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <reporting>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <configuration>
                    <scalaVersion>2.11.8</scalaVersion>
                </configuration>
            </plugin>
        </plugins>
    </reporting>
</project>

2.创建com.test.bigdata 包，并在包下创建 WordCount.scala，代码编写如下。

package com.test.bigdata

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

object WordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("yarn").setAppName("WordCount");
    val sc = new SparkContext(conf)
    val lines = sc.textFile(args(0)); // 0参数指定读取文件
    val line = lines.flatMap(_.split(" ")); // 对每行数据按空格进行切分
    val words = line.map(word => (word, 1)); // 对每个单词计数为1
    val result = words.reduceByKey((x, y) => x + y); // 对单词进行累加统计
    result.saveAsTextFile(args(1)); // 1参数指定输出结果位置
  }
}

3.对项目进行打包，点击Lifecycle>package即可开始打包。

打包成功后可以在target目录下看到生成的jar包，将该jar包移至D盘等待上传运行。

2、上传程序jar包

1.使用PuTTY连接Master节点，使用mkdir命令在该节点创建一个/test目录。

创建文件夹test；

mkdir /test

切换到test路径下；

cd /test

2.找到PuTTY的安装目录，在上方地址栏输入cmd并执行。

3.上传jar包

在弹出的黑窗口首先输入psftp，打开psftp工具用来传输文件；

psftp

接下来连接服务器，回车后需要输入用户名和密码；

open xxx.xxx.xxx.xxx 

用于切换远程Linux 服务器上的目录；

cd /test/

lcd命令用于切换本地的路径，切换至包所在的本地路径D盘；

lcd D:\

上传文件；

put TestSpark-1.0-SNAPSHOT.jar 

命令使用可以参考下图：

3、程序运行与验证

1.下载数据集words.dat到/test目录下

wget https://course-public-resources-1252758970.cos.ap-chengdu.myqcloud.com/%E5%AE%9E%E6%88%98%E8%AF%BE/202001bigdata/7-spark/words.dat

2.上传数据集到HDFS

切换至hadoop用户；

su hadoop

字HDFS中创建一个/test文件夹；

hdfs dfs -mkdir /test

上传words.dat文件至hdfs的/test目录下（执行时没有任何输出，图示输出仅为日志配置信息）；

hdfs dfs -put /test/words.dat /test

3.Spark客户端运行程序，以client模式提交任务，命令的参数说明如下：

/usr/local/service/spark/bin/spark-submit  \
  --class com.test.bigdata.WordCount \
  --master yarn \
  --deploy-mode cluster \
  TestSpark-1.0-SNAPSHOT.jar /test/words.dat /test/output2

4.查看结果。

hdfs dfs -ls /test/output2

可以通过tail命令查看输出文件的部分内容；

hdfs dfs -tail /test/output2/part-00000

命令执行后显示结果如下：

4、查看任务运行记录

1.在弹性MapReduce控制台找到组件管理，点击YARN原生WebUI访问地址。第一次访问需要输入用户名和密码。

2.进入YARN页面后可以查看Spark程序的名称，运行时间，程序运行状态等信息。

至此，您已完成了实验的全部任务。相信您已经掌握了Spark的基本开发和部署流程，您还可以参考官方文档，体验Spark或Hadoop的其他功能。

5. 注意事项

如实验资源无需保留，请在实验结束后及时销毁，以免产生额外费用。

6. FAQ

【问题】1、在任务2中，Maven下载依赖失败。

【解决】学员如果下载依赖失败，可以使用下面接连下载依赖。

Spark项目所需依赖下载

Spark实验jar包参考

下载依赖包解压后，复制到Maven的本地仓库下（粘贴选择合并文件夹，同名文件选择覆盖原有文件）。