spark系列3：spark入门编程与介绍

比眉伴天荒 2022-11-09 12:43 188阅读 0赞

## 3. Spark 入门 ##

目标

1.  通过理解 Spark 小案例, 来理解 Spark 应用
2.  理解编写 Spark 程序的两种常见方式
    
    1.  spark-shell
    2.  spark-submit

Spark 官方提供了两种方式编写代码, 都比较重要, 分别如下

*  `spark-shell`  
    Spark shell 是 Spark 提供的一个基于 Scala 语言的交互式解释器, 类似于 Scala 提供的交互式解释器, Spark shell 也可以直接在 Shell 中编写代码执行  
    这种方式也比较重要, 因为一般的数据分析任务可能需要探索着进行, 不是一蹴而就的, 使用 Spark shell 先进行探索, 当代码稳定以后, 使用独立应用的方式来提交任务, 这样是一个比较常见的流程
 *  `spark-submit`  
    Spark submit 是一个命令, 用于提交 Scala 编写的基于 Spark 框架, 这种提交方式常用作于在集群中运行任务

### 3.1. Spark shell 的方式编写 WordCount ###

概要

在初始阶段工作可以全部使用 Spark shell 完成, 它可以加快原型开发, 使得迭代更快, 很快就能看到想法的结果. 但是随着项目规模越来越大, 这种方式不利于代码维护, 所以可以编写独立应用. 一般情况下, 在探索阶段使用 Spark shell, 在最终使用独立应用的方式编写代码并使用 Maven 打包上线运行

接下来使用 Spark shell 的方式编写一个 WordCount

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>Spark shell 简介</p> 
    <ul> 
     <li> <p>启动 Spark shell<br> 进入 Spark 安装目录后执行&nbsp;<code>spark-shell --master master</code>&nbsp;就可以提交Spark 任务</p> </li> 
     <li> <p>Spark shell 的原理是把每一行 Scala 代码编译成类, 最终交由 Spark 执行</p> </li> 
    </ul></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>Master地址的设置</p> <p>Master 的地址可以有如下几种设置方式</p> 
    <table>  Table 3. master  
     <thead> 
      <tr> 
       <th>地址</th> 
       <th>解释</th> 
      </tr> 
     </thead> 
     <tbody> 
      <tr> 
       <td> <p><code>local[N]</code></p> </td> 
       <td> <p>使用 N 条 Worker 线程在本地运行</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>spark://host:port</code></p> </td> 
       <td> <p>在 Spark standalone 中运行, 指定 Spark 集群的 Master 地址, 端口默认为 7077</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>mesos://host:port</code></p> </td> 
       <td> <p>在 Apache Mesos 中运行, 指定 Mesos 的地址</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>yarn</code></p> </td> 
       <td> <p>在 Yarn 中运行, Yarn 的地址由环境变量&nbsp;<code>HADOOP_CONF_DIR</code>&nbsp;来指定</p> </td> 
      </tr> 
     </tbody> 
    </table></td> 
  </tr> 
 </tbody> 
</table>

Step 1 准备文件

在 Node01 中创建文件 `/export/data/wordcount.txt`

hadoop spark flume
    spark hadoop
    flume hadoop

Step 2 启动 Spark shell

cd /export/servers/spark
    bin/spark-shell --master local[2]

Step 3 执行如下代码

scala> val sourceRdd = sc.textFile("file:///export/data/wordcount.txt")
    sourceRdd: org.apache.spark.rdd.RDD[String] = file:///export/data/wordcount.txt MapPartitionsRDD[1] at textFile at <console>:24
    
    scala> val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_, 1))
    flattenCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:26
    
    scala> val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
    aggCountRdd: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:28
    
    scala> val result = aggCountRdd.collect
    result: Array[(String, Int)] = Array((spark,2), (hadoop,3), (flume,2))

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>sc</p> <p>上述代码中&nbsp;<code>sc</code>&nbsp;变量指的是 SparkContext, 是 Spark 程序的上下文和入口</p> <p>正常情况下我们需要自己创建, 但是如果使用 Spark shell 的话, Spark shell 会帮助我们创建, 并且以变量&nbsp;<code>sc</code>&nbsp;的形式提供给我们调用</p> </td> 
  </tr> 
 </tbody> 
</table>

运行流程

![60a2714b057c19957908cfda93b8c321][]

1.  `flatMap(_.split(" "))` 将数据转为数组的形式, 并展平为多个数据
2.  `map_, 1` 将数据转换为元组的形式
3.  `reduceByKey(_ + _)` 计算每个 Key 出现的次数

总结

1.  使用 Spark shell 可以快速验证想法
2.  Spark 框架下的代码非常类似 Scala 的函数式调用

### 3.2. 读取 HDFS 上的文件 ###

目标

1.  理解 Spark 访问 HDFS 的两种方式

Step 1 上传文件到 HDFS 中

cd /export/data
    hdfs dfs -mkdir /dataset
    hdfs dfs -put wordcount.txt /dataset/

Step 2 在 Spark shell 中访问 HDFS

val sourceRdd = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")
    val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_, 1))
    val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
    val result = aggCountRdd.collect

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>如何使得 Spark 可以访问 HDFS?</p> <p>可以通过指定 HDFS 的 NameNode 地址直接访问, 类似于上面代码中的&nbsp;<code>sc.textFile("hdfs://node01:8020/dataset/wordcount.txt")</code></p> <p><img alt="155c0a820881a7db91ea8d7cc53555d9" src="https://img-blog.csdnimg.cn/img_convert/9075f8fbe8565860a7507b9eeaf48c8d.png"></p> <p>也可以通过向 Spark 配置 Hadoop 的路径, 来通过路径直接访问</p> <p>1.在&nbsp;<code>spark-env.sh</code>&nbsp;中添加 Hadoop 的配置路径</p> <p><code>export HADOOP_CONF_DIR="/etc/hadoop/conf"</code></p> <p>2.在配置过后, 可以直接使用&nbsp;<code>hdfs:///路径</code>&nbsp;的形式直接访问</p> <p><img alt="dd904b1653a52fe15d0bb7808d98b9af" src="https://img-blog.csdnimg.cn/img_convert/6288c929af96113bc3377c7683705f2d.png"></p> <p>3.在配置过后, 也可以直接使用路径访问</p> <p><img alt="3eabed898ed57db55370c25fad555072" src="https://img-blog.csdnimg.cn/img_convert/5dd93b4bdd0e46493221b6a0baf80576.png"></p> </td> 
  </tr> 
 </tbody> 
</table>

### 3.4. 编写独立应用提交 Spark 任务 ###

目标

1.  理解如何编写 Spark 独立应用
2.  理解 WordCount 的代码流程

Step 1 创建工程

1.  创建 IDEA 工程
    
    1.  ![ee1391b4d7e1214b5b4155b6806a6794][] → ![24f103c1662f69cbb0af4bfc8a54b071][] → ![9affa530ce6f4de7da24efa30c5b4227][]
    2.  ![4a8dac7fcd60c730512028265f27699f][] → ![17fd56ce77043ded7754dc08b72a1f63][] → ![412959e49ee5078f2e6d609d14e6307f][]
2.  增加 Scala 支持
    
    1.  右键点击工程目录 ![410a1fe6ae14ce614ee6e50f4e263e51][]
    2.  选择增加框架支持 ![c0c839c6f01db04cc112bfd2af260961][]
    3.  选择 Scala 添加框架支持

Step 2 编写 Maven 配置文件 `pom.xml`

1.  工程根目录下增加文件 `pom.xml`
2.  添加以下内容
    
        <?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>cn.itcast</groupId>
            <artifactId>spark</artifactId>
            <version>0.1.0</version>
        
            <properties>
                <scala.version>2.11.8</scala.version>
                <spark.version>2.2.0</spark.version>
                <slf4j.version>1.7.16</slf4j.version>
                <log4j.version>1.2.17</log4j.version>
            </properties>
        
            <dependencies>
                <dependency>
                    <groupId>org.scala-lang</groupId>
                    <artifactId>scala-library</artifactId>
                    <version>${scala.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.apache.spark</groupId>
                    <artifactId>spark-core_2.11</artifactId>
                    <version>${spark.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.apache.hadoop</groupId>
                    <artifactId>hadoop-client</artifactId>
                    <version>2.6.0</version>
                </dependency>
        
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>jcl-over-slf4j</artifactId>
                    <version>${slf4j.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-api</artifactId>
                    <version>${slf4j.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                    <version>${slf4j.version}</version>
                </dependency>
                <dependency>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                    <version>${log4j.version}</version>
                </dependency>
                <dependency>
                    <groupId>junit</groupId>
                    <artifactId>junit</artifactId>
                    <version>4.10</version>
                    <scope>provided</scope>
                </dependency>
            </dependencies>
        
            <build>
                <sourceDirectory>src/main/scala</sourceDirectory>
                <testSourceDirectory>src/test/scala</testSourceDirectory>
                <plugins>
        
                    <plugin>
                        <groupId>org.apache.maven.plugins</groupId>
                        <artifactId>maven-compiler-plugin</artifactId>
                        <version>3.0</version>
                        <configuration>
                            <source>1.8</source>
                            <target>1.8</target>
                            <encoding>UTF-8</encoding>
                        </configuration>
                    </plugin>
        
                    <plugin>
                        <groupId>net.alchim31.maven</groupId>
                        <artifactId>scala-maven-plugin</artifactId>
                        <version>3.2.0</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>3.1.1</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>
        </project>
3.  因为在 `pom.xml` 中指定了 Scala 的代码目录, 所以创建目录 `src/main/scala` 和目录 `src/test/scala`
4.  创建 Scala object `WordCount`

Step 3 编写代码

object WordCounts {
    
      def main(args: Array[String]): Unit = {
        // 1. 创建 Spark Context
        val conf = new SparkConf().setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
    
        // 2. 读取文件并计算词频
        val source: RDD[String] = sc.textFile("hdfs://node01:8020/dataset/wordcount.txt", 2)
        val words: RDD[String] = source.flatMap { line => line.split(" ") }
        val wordsTuple: RDD[(String, Int)] = words.map { word => (word, 1) }
        val wordsCount: RDD[(String, Int)] = wordsTuple.reduceByKey { (x, y) => x + y }
    
        // 3. 查看执行结果
        println(wordsCount.collect)
      }
    }

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td>和 Spark shell 中不同, 独立应用需要手动创建 SparkContext</td> 
  </tr> 
 </tbody> 
</table>

Step 4 运行

运行 Spark 独立应用大致有两种方式, 一种是直接在 IDEA 中调试, 另一种是可以在提交至 Spark 集群中运行, 而 Spark 又支持多种集群, 不同的集群有不同的运行方式

直接在 IDEA 中运行 Spark 程序

1.  在工程根目录创建文件夹和文件
    
    ![f6ccfd3d52928baa0478100832a723b0][]
2.  修改读取文件的路径为\`dataset/wordcount.txt\`
    
    ![ad2eef5059c8fb5e819d9287c6c9cb25][]
3.  右键运行 Main 方法
    
    ![37b5dcc51939c056608275f89a3d0fc1][]

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>spark-submit 命令</p> <pre><code>spark-submit [options] &lt;app jar&gt; &lt;app options&gt;</code></pre> 
    <ul> 
     <li> <p><code>app jar</code>&nbsp;程序 Jar 包</p> </li> 
     <li> <p><code>app options</code>&nbsp;程序 Main 方法传入的参数</p> </li> 
     <li> <p><code>options</code>&nbsp;提交应用的参数, 可以有如下选项</p> </li> 
    </ul> 
    <table>  Table 4. options 可选参数  
     <tbody> 
      <tr> 
       <td> <p>参数</p> </td> 
       <td> <p>解释</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--master &lt;url&gt;</code></p> </td> 
       <td> <p>同 Spark shell 的 Master, 可以是spark, yarn, mesos, kubernetes等 URL</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--deploy-mode &lt;client or cluster&gt;</code></p> </td> 
       <td> <p>Driver 运行位置, 可选 Client 和 Cluster, 分别对应运行在本地和集群(Worker)中</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--class &lt;class full name&gt;</code></p> </td> 
       <td> <p>Jar 中的 Class, 程序入口</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--jars &lt;dependencies path&gt;</code></p> </td> 
       <td> <p>依赖 Jar 包的位置</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--driver-memory &lt;memory size&gt;</code></p> </td> 
       <td> <p>Driver 程序运行所需要的内存, 默认 512M</p> </td> 
      </tr> 
      <tr> 
       <td> <p><code>--executor-memory &lt;memory size&gt;</code></p> </td> 
       <td> <p>Executor 的内存大小, 默认 1G</p> </td> 
      </tr> 
     </tbody> 
    </table></td> 
  </tr> 
 </tbody> 
</table>

提交到 Spark Standalone 集群中运行

1.  在 IDEA 中使用 Maven 打包
    
    ![adf0a41da23b6c209edd4dc69d8688e6][]
2.  拷贝打包的 Jar 包上传到 node01 中
    
    ![103e4db41405dcf7ba740b4653b5c216][]
3.  在 node01 中 Jar 包所在的目录执行如下命令
    
        spark-submit --master spark://node01:7077 \
        --class cn.itcast.spark.WordCounts \
        original-spark-0.1.0.jar

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td> <p>如何在任意目录执行 spark-submit 命令?</p> 
    <ol> 
     <li> <p>在&nbsp;<code>/etc/profile</code>&nbsp;中写入如下</p> <pre><code>export SPARK_BIN=/export/servers/spark/bin
export PATH=$PATH:$SPARK_BIN</code></pre> </li> 
     <li> <p>执行&nbsp;<code>/etc/profile</code>&nbsp;使得配置生效</p> <pre><code>source /etc/profile</code></pre> </li> 
    </ol></td> 
  </tr> 
 </tbody> 
</table>

总结: 三种不同的运行方式

Spark shell

*  作用
    
     *  一般用作于探索阶段, 通过 Spark shell 快速的探索数据规律
     *  当探索阶段结束后, 代码确定以后, 通过独立应用的形式上线运行
 *  功能
    
     *  Spark shell 可以选择在集群模式下运行, 还是在线程模式下运行
     *  Spark shell 是一个交互式的运行环境, 已经内置好了 SparkContext 和 SparkSession 对象, 可以直接使用
     *  Spark shell 一般运行在集群中安装有 Spark client 的服务器中, 所以可以自有的访问 HDFS

本地运行

*  作用
    
     *  在编写独立应用的时候, 每次都要提交到集群中还是不方便, 另外很多时候需要调试程序, 所以在 IDEA 中直接运行会比较方便, 无需打包上传了
 *  功能
    
     *  因为本地运行一般是在开发者的机器中运行, 而不是集群中, 所以很难直接使用 HDFS 等集群服务, 需要做一些本地配置, 用的比较少
     *  需要手动创建 SparkContext

集群运行

*  作用
    
     *  正式环境下比较多见, 独立应用编写好以后, 打包上传到集群中, 使用\`spark-submit\`来运行, 可以完整的使用集群资源
 *  功能
    
     *  同时在集群中通过\`spark-submit\`来运行程序也可以选择是用线程模式还是集群模式
     *  集群中运行是全功能的, HDFS 的访问, Hive 的访问都比较方便
     *  需要手动创建 SparkContext

[60a2714b057c19957908cfda93b8c321]: /images/20221023/149d48d201f541e7818d4f73fb82ed7b.png
[ee1391b4d7e1214b5b4155b6806a6794]: https://img-blog.csdnimg.cn/img_convert/56d1a46a7faf27e86999c3ceb35dc6cc.png
[24f103c1662f69cbb0af4bfc8a54b071]: https://img-blog.csdnimg.cn/img_convert/4d2e1a660795f8c0d872fcb7775b06a9.png
[9affa530ce6f4de7da24efa30c5b4227]: /images/20221023/b38302a110ba4e34b920e8f16b99d27b.png
[4a8dac7fcd60c730512028265f27699f]: https://img-blog.csdnimg.cn/img_convert/d7e90c947a37a3b6c3515f8778b5e32f.png
[17fd56ce77043ded7754dc08b72a1f63]: https://img-blog.csdnimg.cn/img_convert/93f5a4c9ab77bc7227d5b7030641d194.png
[412959e49ee5078f2e6d609d14e6307f]: /images/20221023/e0f2f782548c445fb0ba811332bcab77.png
[410a1fe6ae14ce614ee6e50f4e263e51]: /images/20221023/f4e923eb158442bc8a8b79924cc96e43.png
[c0c839c6f01db04cc112bfd2af260961]: /images/20221023/2e5a6171178c462eab9b235406ba91bb.png
[f6ccfd3d52928baa0478100832a723b0]: /images/20221023/3ce47f822c11495a9aaf05a927c8ba63.png
[ad2eef5059c8fb5e819d9287c6c9cb25]: /images/20221023/b812e39e4cf64254bf1c8c0c87304540.png
[37b5dcc51939c056608275f89a3d0fc1]: /images/20221023/3cf17afc80954e40b73c20865ce97bb7.png
[adf0a41da23b6c209edd4dc69d8688e6]: /images/20221023/16e6bea73a0649518c2adb5d62f1bc40.png
[103e4db41405dcf7ba740b4653b5c216]: /images/20221023/8888007af15d45878d2ab49bdaa5f748.png