[Spark]Spark RDD 指南三 弹性分布式数据集(RDD)

简介: Spark2.3.0 版本: Spark2.3.0 创建RDDSpark的核心概念是弹性分布式数据集(RDD),RDD是一个可容错、可并行操作的分布式元素集合。

Spark2.3.0 版本Spark2.3.0 创建RDD

Spark的核心概念是弹性分布式数据集(RDD),RDD是一个可容错、可并行操作的分布式元素集合。有两种方法可以创建RDD对象:

  • 在驱动程序中并行化操作集合对象来创建RDD
  • 从外部存储系统中引用数据集(如:共享文件系统、HDFS、HBase或者其他Hadoop支持的数据源)。

1. 并行化集合

通过在驱动程序中的现有集合上调用JavaSparkContextparallelize方法创建并行化集合(Parallelized collections)。集合的元素被复制以形成可以并行操作的分布式数据集。 例如,下面是如何创建一个包含数字1到5的并行化集合:

Java版本:

List<Integer> list = Arrays.asList(1,2,3,4,5);
JavaRDD<Integer> rdd = sc.parallelize(list);

Scala版本:

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

Python版本:

data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)

RDD一旦创建,分布式数据集(distData)可以并行操作。 例如,我们可以调用distData.reduce((a,b) - > a + b)来实现对列表元素求和。 我们稍后介绍分布式数据集的操作。

并行化集合的一个重要参数是分区(partition)数,将分布式数据集分割成多少分区。Spark集群中每个分区运行一个任务(task)。典型场景下,一般为每个CPU分配2-4个分区。但通常而言,Spark会根据你集群的状况,自动设置分区数。当然,你可以给parallelize方法传递第二个参数来手动设置分区数(如:sc.parallelize(data, 10))。注意:Spark代码里有些地方仍然使用分片(slice)这个术语(分区的同义词),主要为了保持向后兼容。

2. 外部数据集

Spark可以从Hadoop支持的任何存储数据源创建分布式数据集,包括本地文件系统,HDFS,Cassandra,HBase,Amazon S3等。Spark可以支持文本文件,SequenceFiles以及任何其他Hadoop 输入格式。

文本文件RDD可以使用SparkContexttextFile方法创建。该方法根据URL获取文件(机器上的本地路径,或hdfs://s3n://等等). 下面是一个示例调用:

Java版本:

JavaRDD<String> distFile = sc.textFile("data.txt");

Scala版本:

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

Python版本:

>>> distFile = sc.textFile("data.txt")

一旦创建完成,distFiile 就能做数据集操作。例如,我们可以用下面的方式使用 map 和 reduce 操作将所有行的长度相加:

distFile.map(s -> s.length()).reduce((a, b) -> a + b);

Spark读文件时一些注意事项:

(1) 如果使用本地文件系统路径,在所有工作节点上该文件必须都能用相同的路径访问到。要么复制文件到所有的工作节点,要么使用网络的方式共享文件系统。

(2) Spark 所有基于文件的输入方法,包括 textFile,能很好地支持文件目录,压缩过的文件和通配符。例如,你可以使用:

textFile("/my/directory")
textFile("/my/directory/*.txt")
textFile("/my/directory/*.gz")

(3) textFile 方法也可以选择第二个可选参数来控制文件分区(partitions)数目,默认情况下,Spark为每一个文件块创建一个分区(HDFS中分块大小默认为128MB),你也可以通过传递一个较大数值来请求更多分区。 注意的是,分区数目不能少于分块数目。

除了文本文件,Spark的Java API还支持其他几种数据格式:

(1) JavaSparkContext.wholeTextFiles可以读取包含多个小文本文件的目录,并将它们以(文件名,内容)键值对返回。 这与textFile相反,textFile将在每个文件中每行返回一条记录。

JavaPairRDD<String, String> rdd = sc.wholeTextFiles("/home/xiaosi/wholeText");
List<Tuple2<String, String>> list = rdd.collect();
for (Tuple2<?, ?> tuple : list) {
    System.out.println(tuple._1() + ": " + tuple._2());
}

(2) 对于SequenceFiles,可以使用SparkContext的sequenceFile [K,V]方法,其中K和V是文件中的键和值的类型。 这些应该是Hadoop的Writable接口的子类,如IntWritable和Text。

(3) 对于其他Hadoop InputFormats,您可以使用JavaSparkContext.hadoopRDD方法,该方法采用任意JobConf和输入格式类,键类和值类。 将这些设置与使用输入源的Hadoop作业相同。 您还可以使用基于“新”MapReduce API(org.apache.hadoop.mapreduce)的InputFormats的JavaSparkContext.newAPIHadoopRDD。

(4) JavaRDD.saveAsObjectFile 和 SparkContext.objectFile 支持保存一个RDD,保存格式是一个简单的 Java 对象序列化格式。这是一种效率不高的专有格式,如 Avro,它提供了简单的方法来保存任何一个 RDD。

原文:http://spark.apache.org/docs/latest/programming-guide.html#resilient-distributed-datasets-rdds

目录
相关文章
|
7月前
|
分布式计算 Hadoop 大数据
大数据技术与Python:结合Spark和Hadoop进行分布式计算
【4月更文挑战第12天】本文介绍了大数据技术及其4V特性,阐述了Hadoop和Spark在大数据处理中的作用。Hadoop提供分布式文件系统和MapReduce,Spark则为内存计算提供快速处理能力。通过Python结合Spark和Hadoop,可在分布式环境中进行数据处理和分析。文章详细讲解了如何配置Python环境、安装Spark和Hadoop,以及使用Python编写和提交代码到集群进行计算。掌握这些技能有助于应对大数据挑战。
749 1
|
1月前
|
分布式计算 Java 开发工具
阿里云MaxCompute-XGBoost on Spark 极限梯度提升算法的分布式训练与模型持久化oss的实现与代码浅析
本文介绍了XGBoost在MaxCompute+OSS架构下模型持久化遇到的问题及其解决方案。首先简要介绍了XGBoost的特点和应用场景,随后详细描述了客户在将XGBoost on Spark任务从HDFS迁移到OSS时遇到的异常情况。通过分析异常堆栈和源代码,发现使用的`nativeBooster.saveModel`方法不支持OSS路径,而使用`write.overwrite().save`方法则能成功保存模型。最后提供了完整的Scala代码示例、Maven配置和提交命令,帮助用户顺利迁移模型存储路径。
|
4月前
|
存储 分布式计算 数据处理
解释弹性分布式数据集(RDD)的概念
【8月更文挑战第13天】
210 4
|
4月前
|
UED 存储 数据管理
深度解析 Uno Platform 离线状态处理技巧:从网络检测到本地存储同步,全方位提升跨平台应用在无网环境下的用户体验与数据管理策略
【8月更文挑战第31天】处理离线状态下的用户体验是现代应用开发的关键。本文通过在线笔记应用案例,介绍如何使用 Uno Platform 优雅地应对离线状态。首先,利用 `NetworkInformation` 类检测网络状态;其次,使用 SQLite 实现离线存储;然后,在网络恢复时同步数据;最后,通过 UI 反馈提升用户体验。
108 0
|
7月前
|
分布式计算 Hadoop 大数据
分布式计算框架比较:Hadoop、Spark 与 Flink
【5月更文挑战第31天】Hadoop是大数据处理的开创性框架,专注于大规模批量数据处理,具有高扩展性和容错性。然而,它在实时任务上表现不足。以下是一个简单的Hadoop MapReduce的WordCount程序示例,展示如何统计文本中单词出现次数。
223 0
|
7月前
|
SQL 分布式计算 Hadoop
Spark分布式内存计算框架
Spark分布式内存计算框架
198 0
|
7月前
|
机器学习/深度学习 负载均衡 PyTorch
PyTorch分布式训练:加速大规模数据集的处理
【4月更文挑战第18天】PyTorch分布式训练加速大规模数据集处理,通过数据并行和模型并行提升训练效率。`torch.distributed`提供底层IPC与同步,适合定制化需求;`DistributedDataParallel`则简化并行过程。实际应用注意数据划分、通信开销、负载均衡及错误处理。借助PyTorch分布式工具,可高效应对深度学习的计算挑战,未来潜力无限。
|
7月前
|
存储 缓存 分布式计算
Spark【基础知识 02】【弹性式数据集RDDs】(部分图片来源于网络)
【2月更文挑战第13天】Spark【基础知识 02】【弹性式数据集RDDs】(部分图片来源于网络)
84 1
|
7月前
|
分布式计算 大数据 数据处理
Spark RDD(弹性分布式数据集)
Spark RDD(弹性分布式数据集)