当前位置：首页 > 编程日记 > 正文

Spark入门系列（二）| 1小时学会RDD编程

编程日记 2024-12-29 14:10:00

640?wx_fmt=jpeg

作者 | 梁云1991

转载自Python与算法之美（ID:Python_Ai_Road）

导读：本文为 Spark入门系列的第二篇文章，主要介绍 RDD 编程，实操性较强，感兴趣的同学可以动手实现一下。
RDD 是弹性分布式数据集(Resilient Distributed Dataset)，是 Spark 对数据的核心抽象。RDD 其实是分布式的元素集合，当 Spark 对数据操作和转换时，会自动将RDD中的数据分发到集群，并将操作并行化执行。每个 RDD 都被分为多个分区，这些分区运行在集群中的不同节点。

一、编程环境

以下为Mac系统上单机版Spark练习编程环境的配置方法。

注意：仅配置练习环境无需安装Hadoop，无需安装Scala。

1，安装Java8

注意避免安装其它版本的jdk，否则会有不兼容问题。

https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

2，下载spark并解压

http://spark.apache.org/downloads.html

解压到以下路径：

Users/yourname/ProgramFiles/spark-2.4.3-bin-hadoop2.7

3，配置spark环境

vim ~/.bashrc

插入下面两条语句

export SPARK_HOME=/Users/yourname/ProgramFiles/spark-2.4.3-bin-hadoop2.7	
export PATH=$PATH:$SPARK_HOME/bin

4，配置jupyter支持

若未有安装jupyter可以下载Anaconda安装之。使用toree可以安装jupyter环境下的Apache Toree-Scala内核，以便在jupyter环境下运行Spark。

pip install toree	
jupyter toree install --spark_home=Users/yourname/ProgramFiles/spark-2.4.3-bin-hadoop2.7

二、运行Spark

Spark可以通过以下一些方式运行。

1，通过spark-shell进入Spark交互式环境，使用Scala语言。

2，通过spark-submit提交Spark应用程序进行批处理。
这种方式可以提交Scala或Java语言编写的代码编译后生成的jar包，也可以直接提交Python脚本。

3，通过pyspark进入pyspark交互式环境，使用Python语言。
这种方式可以指定jupyter或者ipython为交互环境。

4，通过zepplin notebook交互式执行。
zepplin是jupyter notebook的apache对应产品。

5，安装Apache Toree-Scala内核。
可以在jupyter 中运行spark-shell。

使用spark-shell运行时，还可以添加两个常用的两个参数。

一个是master指定使用何种分布类型。

第二个是jars指定依赖的jar包。

#local本地模式运行，默认使用4个逻辑CPU内核	
spark-shell	#local本地模式运行，使用全部内核，添加 code.jar到classpath	
spark-shell  --master local[*] --jars code.jar	#local本地模式运行，使用4个内核	
spark-shell  --master local[4]	#standalone模式连接集群，指定url和端口号	
spark-shell  --master spark://master:7077	#客户端模式连接YARN集群，Driver运行在本地，方便查看日志，调试时推荐使用。	
spark-shell  --master yarn-client	#集群模式连接YARN集群，Driver运行在集群，本地机器计算和通信压力小，批量任务时推荐使用。	
spark-shell  --master yarn-cluster

#提交scala写的任务	
./bin/spark-submit --class org.apache.spark.examples.SparkPi \	--master yarn \	--deploy-mode cluster \	--driver-memory 4g \	--executor-memory 2g \	--executor-cores 1 \	--queue thequeue \	examples/jars/spark-examples*.jar 10

#提交python写的任务	
spark-submit --master yarn \	
--executor-memory 6G \	
--driver-memory 6G \	
--deploy-mode cluster \	
--num-executors 600 \	
--conf spark.yarn.maxAppAttempts=1 \	
--executor-cores 1 \	
--conf spark.default.parallelism=2000 \	
--conf spark.task.maxFailures=10 \	
--conf spark.stage.maxConsecutiveAttempts=10 \	
test.py

三、创建RDD

创建RDD的基本方式有两种，第一种是使用textFile加载本地或者集群文件系统中的数据。第二种是使用parallelize方法将Driver中的数据结构并行化成RDD。

1，textFile

640?wx_fmt=png

2，parallelize（或makeRDD）

640?wx_fmt=png

四、常用Action操作

Action操作将触发基于RDD依赖关系的计算。

1，collect

640?wx_fmt=png

2，take

640?wx_fmt=png

3，takeSample

640?wx_fmt=png

4，first

640?wx_fmt=png

5，count

640?wx_fmt=png

6，reduce

640?wx_fmt=png

7，foreach

640?wx_fmt=png

8，coutByKey

640?wx_fmt=png

9，saveAsFile

640?wx_fmt=png

五、常用Transformation操作

Transformation转换操作具有懒惰执行的特性，它只指定新的RDD和其父RDD的依赖关系，只有当Action操作触发到该依赖的时候，它才被计算。

1，map

640?wx_fmt=png

2，filter

640?wx_fmt=png

3，flatMap

640?wx_fmt=png

4，sample

640?wx_fmt=png

5，distinct

640?wx_fmt=png

6，subtract

640?wx_fmt=png

7，union

640?wx_fmt=png

8，intersection

640?wx_fmt=png

9，cartesian

640?wx_fmt=png

10，sortBy

640?wx_fmt=png

11，pipe

640?wx_fmt=png

六、常用PairRDD转换操作

PairRDD指的是数据为Tuple2数据类型的RDD，其每个数据的第一个元素被当做key，第二个元素被当做value。

1，reduceByKey

640?wx_fmt=png

2，groupByKey

640?wx_fmt=png

3，sortByKey

640?wx_fmt=png

4，join

640?wx_fmt=png

5，leftOuterJoin

640?wx_fmt=png

6，rightOuterJoin

640?wx_fmt=png

7，cogroup

640?wx_fmt=png

8，subtractByKey

640?wx_fmt=png

9，foldByKey

640?wx_fmt=png

七、持久化操作

如果一个RDD被多个任务用作中间量，那么对其进行cache，缓存到内存中会对加快计算非常有帮助。

声明对一个RDD进行cache后，该RDD不会被立即缓存，而是等到它第一次因为某个Action操作触发后被计算出来时才进行缓存。

可以使用persist明确指定存储级别，常用的存储级别是MEMORY_ONLY和MEMORY_AND_DISK。

1，cache

640?wx_fmt=png