0


Spark编程基础

一,Spark设计与运行原理

1,Spark简介

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

特点:运行速度快、容易使用、通用性、运行模式多样

2,Spark与Hadoop对比

Hadoop存在的缺点:

表达能力有限 ,磁盘IO开销大 ,延迟高

Spark优点:

编程模型更灵活,迭代运算效率更高,任务调度机制更优

二,Spark运行架构

1,RDD设计与运行原理

创建RDD

方法1.parallelize()

parallelizeO方法有两个输人参数,说明如下:
(1)要转化的集合:必须是 Seq集合。Seq 表示序列,指的是一类具有一定长度的、可迭代访问的对象,其中每个数据元素均带有一个从0开始的、固定的索引。
(2)分区数。若不设分区数,则RDD 的分区数默认为该程序分配到的资源的 CPU核心数。
通过 parallelizeO方法用一个数组的数据创建RDD,并设置分区数为4,创建后查看该 RDD 的分区数

方法2.makeRDD()

makeRDD0方法有两种使用方式,第一种使用方式与 parallelize0方法一致;第二种方式是通过接收一个 Seq[(T,Seq[String])]参数类型创建 RDD。第二种方式生成的RDD中保存的是T的值,Seq[String]部分的数据会按照 Seqf(T,Seq[String])的顺序存放到各个分区中,一个 Seq[Stringl对应存放至一个分区,并为数据提供位置信息,通过preferredLocations0方法可以根据位置信息查看每一个分区的值。调用 makeRDD0时不可以直接指定 RDD 的分区个数,分区的个数与 Seq[String]参数的个数是保持一致的,使用 makeRDD0方法创建 RDD,并根据位置信息查看每一个分区的值

方法3.通过HDFS文件创建 RDD

这种方式较为简单和常用,直接通过 textFile()方法读取 HDFS文件的位置即可。
在HDFS 的/user/toot 目录下有一个文件test.txt,读取该文件创建一个 RDD

方法4.通过 Linux 本地文件创建 RDD

本地文件的读取也是通过 sc.textFile("路径")的方法实现的,在路径前面加上“file://”表示从Linux 本地文件系统读取。在 IntelliJIDEA 开发环境中可以直接读取本地文件;但在 spark-shell 中,要求在所有节点的相同位置保存该文件才可以读取它,例如,在Linux的/opt 目录下创建一个文件 test.txt,任意输入4行数据并保存,将 test.txt 文件远程传输至所有节点的/opt 目录下,才可以读取文件 test.txt。读取 test.txt 文件,并且统计文件的数据行数

2,RDD方法归纳

1.使用map()方法转换数据

2.使用 sortBy()方法进行排序

3.使用flatMap()方法转换数据

**4.使用take()**方法查询某几个值

5.使用****union()方法合并多个RDD

**6.使用distinct()**方法进行去重

3, 使用简单的集合操作

(1)intersection()方法

(2)subtract()方法

(3)**cartesian()**方法

任务实现:

三,Spark快速上手

1,创建Maven项目

(1)增加Scala插件

(2)增加依赖关系

修改Maven项目中的POM文件,增加Spark框架的依赖关系
<dependencies>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-core_2.12</artifactId>

            <version>2.4.5</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <!-- 该插件用于将 Scala 代码编译成 class 文件 -->

            <plugin>

                <groupId>net.alchim31.maven</groupId>

                <artifactId>scala-maven-plugin</artifactId>

                <version>3.2.2</version>

                <executions>

                    <execution>

                        <!-- 声明绑定到 maven 的 compile 阶段 -->

                        <goals>

                            <goal>testCompile</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-assembly-plugin</artifactId>

                <version>3.1.0</version>

                <configuration>

                    <descriptorRefs>

                        <descriptorRef>jar-with-dependencies</descriptorRef>

                    </descriptorRefs>

                </configuration>

                <executions>

                    <execution>

                        <id>make-assembly</id>

                        <phase>package</phase>

                        <goals>

                            <goal>single</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>

** (3)WordCount**

为了能直观地感受Spark框架的效果,接下来我们实现一个大数据学科中最常见的教学案例 WordCount

/**
 * spark实现单词计数
 */
 
object WordCountSpark {
 
  def main(args: Array[String]): Unit = {
 
    //创建spark运行配置对象
 
    val spark: SparkConf = new SparkConf()
 
      .setMaster("local[*]")
 
      .setAppName("WordCountSparkApps")
 
    //创建spark上下文对象
 
    val sc: SparkContext = new SparkContext(spark)
 
    //读文件数据
 
    val wordsRDD: RDD[String] = sc.textFile("data/word.txt")
 
    //讲文件中的数据进行分词
 
    val word: RDD[String] = wordsRDD.flatMap(_.split(","))
 
    //转换数据结构word ---->(word,1)
 
    val word2: RDD[(String, Int)] = word.map((_, 1))
 
    //将转换结构后的数据按照相同的单词进行分组聚合
 
    val word2CountRDD: RDD[(String, Int)] = word2.reduceByKey(_ + _)
 
    //将数据聚合结果采集到内存中
 
    val word2Count: Array[(String, Int)] = word2CountRDD.collect()
 
    //打印结果
 
    word2Count.foreach(println)
 
    //关闭spark连接
 
    sc.stop()
 
  }
 
}

(4)异常处理

如果本机操作系统是Windows,在程序中使用了Hadoop相关的东西,比如写入文件到HDFS,则会遇到如下异常:

出现这个问题的原因,并不是程序的错误,而是windows系统用到了hadoop相关的服务,解决办法是通过配置关联到windows的系统依赖就可以了

在IDEA中配置RunConfiguration,添加HADOOP_HOME变量或者在windows上配置环境变量:

四,Spark运行环境

1,Local模式

上传并解压缩文件

(1)上传文件至/usr/local/packages中

(2)解压缩到指定目录

[root@master local]# tar -zxvf spark-2.4.5-bin-hadoop2.6.tgz -C /usr/local/soft/

(3)重命名

[root@master soft]# mv spark-2.4.5-bin-hadoop2.6/ spark-local

启动local环境

(1)进入解压缩(spark-local)目录

(2)启动成功后,可以输入网址进行Web UI监控页面访问

命令行工具

sc.textFile("data/word.txt").flatMap(.split(",")).map((,1)).reduceByKey(+).collect

退出本地模式

scala> :quit

提交应用

bin/spark-submit \

--class org.apache.spark.examples.SparkPi \

--master local[1] \

./examples/jars/spark-examples_2.11-2.4.5.jar \

10

五,Spark运行架构

1,运行架构

Spark框架的核心是一个计算引擎,整体来说,它采用了标准master-slave的结构。如下图所示,它展示了一个 Spark 执行时的基本结构。图形中的Driver表示 master,负责管理整个集群中的作业任务调度。图形中的Executor则是 slave,负责实际执行任务。

2,核心概念

(1)Executor与Core


本文转载自: https://blog.csdn.net/Xiaozou_Q/article/details/138896913
版权归原作者 Xiaozou_Q 所有, 如有侵权,请联系我们删除。

“Spark编程基础”的评论:

还没有评论