【spark 深入学习 05】RDD编程之旅基础篇-01

拼搏现实的明天。 2021-12-24 11:41 263阅读 0赞

**----------------**

本节内容

**1.RDD****的工作流程**

**2.WordCount****解说**

**· shell版本WordCount**

**· java版本WordCount**

**----------------**

**一、RDD工作流程**

1. RDD是spark特有的数据模型，谈到RDD就会提到什么弹性分布式数据集，什么有向无环图，本文暂时不去展开这些高深概念，在阅读本文时候，大家可以就把RDD当作一个数组，这样的理解对我们学习RDD的API是非常有帮助的。本文所有示例代码都是使用scala语言编写的。RDD的执行过程如下：

· 从外部数据创建出输入RDD,或者从驱动程序分发驱动程序中的对象集合

· 对RDD进行转化，一个RDD转化为一个新的RDD，如filter()操作

· 如果需要重用，告知RDD执行persist()操作

· 执行action触发计算并行计算，spark先优化再执行计算，如count()和first()

RDD的创建有2种方式

(1)从驱动程序分发驱动程序中的对象集合  
从内存里构造RDD，使用的方法：makeRDD和parallelize方法

\-----------------------

val rdd01 = sc.makeRDD(List(1,2,3,4,5,6));

val r01 = rdd01.map \{ x => x \* x \}

println(r01.collect().mkString(","))

/\* Array \*/

val rdd02 = sc.makeRDD(Array(1,2,3,4,5,6))

val r02 = rdd02.filter \{ x => x < 5\}

println(r02.collect().mkString(","))

val rdd03 = sc.parallelize(List(1,2,3,4,5,6), 1)

val r03 = rdd03.map \{ x => x + 1 \}

println(r03.collect().mkString(","))

/\* Array \*/

val rdd04 = sc.parallelize(Array(1,2,3,4,5,6) ,1)

val r04 = rdd04.filter \{ x => x > 3 \}

println(r04.collect().mkString(","))

\-----------------------

2.makeRDD和parallelize的区别

makeRDD有两种实现方式，第一种方式parallelize声明都一样，接收的参数和parallelize完全一样，def makeRDD\[T:ClassTag\]，这种实现方式的makeRDD依赖了parallelize；makeRDD第二种实现方式defmakeRDD\[T:ClassTag\](T,Seq(String)))

第一种:mkRDD实现方式

val blog1=sc.parallelize(List(1,2,3));

val blog2=sc.makeRDD(List(1,2,3));

第二种:mkRDD实现方式

valseq=List((1,List("a","b","c")),(2,List("aa","bb","cc")));

val blog3=sc.makeRDD(seq);

blog3.preferredLocations(blog3.partitions(0));

blog3.preferredLocations(blog3.partitions(1));

**WordCount**

WordCount是分布式编程的入门示例，本节也从WordCount举例说明RDD DEMO

1.Spark shell版本

\---------------------------------------------------

//加载hdfs上的文件

val txtFile ="/tmp/test/core-site.xml" ;

val txtData = sc.textFile(txtFile);

//将上一步生成的RDD对象保存到缓存中，在此之后Spark就不需要在每次数据查询时都重新计算

txtData.cache() ;

// flatMap先映射后扁平化,

val wcData = txtData.flatMap(l =>l.split(" ")).map(word => (word, 1)).reduceByKey(\_ + \_);

//可以提取出所有rdd里的数据项,逐行输出

wcData.collect().foreach(println);

备注：

A. 关于spark-shell的启动参数指定

bin/spark-shell --executor-memory 1G --total-executor-cores10 --executor-cores 1 --master yarn-client --driver-class-path /usr/local/tdr\_hadoop/spark/spark-1.6.0-bin-hadoop2.6/lib/mysql-connector-java-5.1.40-bin.jar

\--executor-memory: 指定每个executor(执行器)占用的内存

\--total-executor-cores: 所有executor总共使用的cpu核数

\--executor-cores：每个executor使用的cpu核数

\--driver-class-path:指定要加载的jar包

\--master:

local\[8\]：表示在本地运行，数据会下载到接口机本地来执行，单机版

spark://master01:7077：表示在集群上运行应用程序，指定任务提交的集群路径在哪里。这就需要提前启动一个真实的Standalone集群。可以指定多个master的地址，用逗号隔开。

yarn-client：在客户模式上，driver与提交程序的客户端在一个进程

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1][]

yarn-cluster：在集群模式上，driver是从集群中的一个worker进程中启动的，这个进程只要完成了提交作业任务就会退出，不会等待提交的应用程序的完成。Spark-shell时，必须使用yarn-client模式，因为你要在client上写SQL。

B.spark-shell 是一个spark application，运行时需要向资源管理器申请资源，如standalone spark、YARN、Mesos。本例向standalone spark申请资源，所以在运行spark-shell时需要指向申请资源的standalonespark集群信息，其参数为MASTER。

如果未在spark-env.sh中申明MASTER，则使用命令MASTER=spark://cdh1:7077bin/spark-shell启动；如果已经在spark-env.sh中申明MASTER，则可以直接用bin/spark-shell启动。

由于spark-shell缺省的情况下，会申请所有的CPU资源

B. Spark每次Executor执行任务情况

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 1][]

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 2][]

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 3][]

2. java 版本

搭建Spark开发环境

（1）前提：配置好jdk和scala到windows  
（2）安装Intellij去官网下载Intellij：https://www.jetbrains.com/idea/，在windows环境下双击安装即可  
（3）安装scala插件

步骤如下图所示，安装好scala插件后，点击restart重启intellij

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 4][]

（4）、使用Intellij写WordCount代码  
a.新建scala工程  
File -> new -> project -> scala project –>scala，项目名称：spark02

在src目录下，建立cn.com包，在该包下建立object 类，命名为word，完成word.scala代码如下所示：

\---------------------------------------------------------------------**package** cn.com  
**import** org.apache.spark.SparkConf  
**import** org.apache.spark.SparkContext  
**import** org.apache.spark.SparkContext.\_  
  
*/\*\*  
\* Created by Administrator on 2016/11/2.  
\*/*  
**object** word \{  
**def** main(args: Array\[String\]) \{  
**if**(args.length < 1) \{  
System.*err*.println(**"Usage: <file>"**)  
System.*exit*(1)  
\}  
**val** conf = **new** SparkConf()  
**val** sc = **new** SparkContext(conf)  
*//SparkContext* *是把代码提交到集群或者本地的通道，我们编写Spark代码，无论是要本地运行还是集群运行都必须有SparkContext的实例*  
**val** line = sc.textFile(args(0))  
*//**把读取的内容保存给line变量，其实line是一个MappedRDD，Spark的所有操作都是基于RDD的*  
line.flatMap(\_.split(**" "**)).map((\_, 1)).reduceByKey(\_+\_).collect.foreach(*println*)  
sc.stop  
\}  
  
\}

\---------------------------------------------------------------------

b.导入spark包  
File

\->Project structure

\->project settting

\->libraries->+  
![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 5][]

导入spark-assembly-1.6.0-hadoop2.6.0.jar包（该包从spark安装包的lib下获得）

c.选择Artifacts

File

\->Project structure

\->project settting

\->Artifacts->+，选择要导入的项目，以及main类

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 6][]

并且指定jar包输出的位置

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 7][]

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 8][]

d．输出jar包  
Build -> Build ArtiFacts ->build,打好jar包到：D:\\spark02\\out\\artifacts\\spark02\_jar\\spark02.jar

e．上传jar包到spark客户端，并执行  
执行命令：

spark-submit --master yarn --executor-memory 1000M /usr/local/tdr\_hadoop/spark/spark02.jarhdfs://tdrHadoop/tmp/test/core-site.xml

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 9][]

在yarn的前台显示正在执行

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 10][]

执行结果输出：

![640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 11][]

转载于:https://www.cnblogs.com/licheng/p/6815309.html

[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1]: /images/20211223/cfce24c9b4cf4fbeb25b062637e09b47.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 1]: /images/20211223/c6e4434fe08a4537ab44a5c9474260db.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 2]: /images/20211223/af39e1450e2946a796421b3f9faad885.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 3]: /images/20211223/3a877e1bea5f45078ed8ff6ba896651b.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 4]: /images/20211223/77e8c6962fa34e5eac9145f8ea93e88d.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 5]: /images/20211223/498aa716594f4be3939b49fe400db6ed.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 6]: /images/20211223/48711c9588ab42a09640c3dc17315989.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 7]: /images/20211223/ac7f598a9d8d47f1a3ea887fdd5adb91.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 8]: /images/20211223/979e8632eab7488baa8bcbf06efc3d4e.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 9]: /images/20211223/23689969442442199bcad4fc766a1383.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 10]: /images/20211223/a053f0919faa4b9c9ad7c02f64e3dd3e.png
[640_wx_fmt_png_tp_webp_wxfrom_5_wx_lazy_1 11]: /images/20211223/d2364d7fe9e4450db1efd441c79bcfba.png