Spark RDD编程初级实践(一）

任务描述

本关任务：编写Spark独立应用程序实现数据去重。
相关知识

为了完成本关任务，你需要掌握：RDD的创建；RDD的转换操作；RDD的行动操作。
RDD的创建

使用textFile()方法从本地文件系统中加载数据创建RDD，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt")执行sc.textFile()方法以后，Spark从本地文件word.txt中加载数据到内存，在内存中生成一个RDD对象lines，这个RDD里面包含了若干个元素，每个元素的类型是String类型，也就是说，从word.txt文件中读取出来的每一行文本内容，都成为RDD中的一个元素。使用map()函数转换得到相应的键值对RDD，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt")val pairRDD = lines.flatMap(line => line.split(" ")).map(word => (word,1))上面示例中，map(word=>(word,1))函数的作用是取出RDD中的每个元素，也就是每个单词，赋值给word，然后把word转换成 (word,1) 的键值对形式。

RDD的转换操作

对于RDD而言，每一次转换操作都会产生新的RDD，供给下一个操作使用。RDD的转换过程是惰性求值的，也就是说，整个转换过程只是记录了转换的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会触发真正的计算。常见的RDD转换操作如下所示：

**filter(func)**：筛选出满足函数func的元素，并返回一个新的RDD，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt")val linesWithSpark = lines.filter(line => line.contains("Spark"))**map(func)**：将每个元素传递到函数func中，并将结果返回为一个新的RDD，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt")val words = lines.map(line => line.split(" "))**groupByKey()**：应用于(K,V)键值对RDD时，返回一个新的(K, Iterable)形式的RDD，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt")val words = lines.flatMap(line => line.split(" ")).map(word => (word,1)).groupByKey()**sortByKey()**：应用于(K,V)键值对RDD时，返回一个新的根据key排序的RDD，示例如下：val pairRDD = sc.parallelize(Array("Hadoop",3),("Spark",5),("Hive",2))pairRDD.sortByKey().foreach(println)输出：
(Hadoop,3)
(Hive,2)
(Spark,5)**partitionBy(partitioner: Partitioner)**：根据partitioner函数生成新的ShuffleRDD，将原RDD重新分区，示例如下：val lines = sc.textFile("file:///home/hadoop/word.txt", 3)val words = lines.flatMap(line => line.split(" ")).map(word => (word,1)).partitionBy(new HashPartitioner(1))keys：将键值对RDD中所有元素的key返回，形成一个新的RDD，示例如下：val pairRDD = sc.parallelize(Array("Hadoop",3),("Spark",5),("Hive",2))pairRDD.keys.foreach(println)输出：
Hadoop
Spark
Hive

RDD的行动操作

对于RDD而言，只有遇到行动操作时才会执行“从头到尾”的真正的计算，从文件中加载数据，完成一次又一次转换操作，最终，完成行动操作得到结果。常见的RDD行动操作如下所示：

**count()**：返回RDD中元素的个数
**collect()**：以数组的形式返回RDD中的所有元素
**first()**：返回RDD中的第一个元素
**take(n)**：以数组的形式返回RDD中的前n个元素
**reduce(func)**：通过函数func(输入两个参数并返回一个值)聚合RDD中的元素
**foreach(func)**：将RDD中的每个元素传递到函数func中运行 下面通过一个示例来介绍上述行动操作，如下所示：val rdd = sc.parallelize(Array(1,2,3,4,5))println(rdd.count)println(rdd.first)println(rdd.take(3).mkString(", "))println(rdd.collect().mkString("-> "))rdd.foreach(println)输出： 5 1 1, 2, 3 1-> 2-> 3-> 4-> 5 1 2 3 4 5

编程要求

对于两个输入文件A和B，编写Spark独立应用程序，对两个文件进行合并、剔除其中重复的内容并排序，得到一个新文件C。下面是输入文件和输出文件的一个样例，供参考。
输入文件A的样例如下：
20170101 x
20170102 y
20170103 x
20170104 y
20170105 z
20170106 z
输入文件B的样例如下：
20170101 y
20170102 y
20170103 x
20170104 z
20170105 y
输出文件C的样例如下：
20170101 x
20170101 y
20170102 y
20170103 x
20170104 y
20170104 z
20170105 y
20170105 z
20170106 z
测试说明

本实训目前是基于Spark单机模式的运行方式，完成整个评测流程所需时间较长（全过程耗时约60秒），请耐心等待!

开始你的任务吧，祝你成功！

代码

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.HashPartitioner
object RemDup {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("RemDup").setMaster("local")val sc = new SparkContext(conf)//输入文件fileA.txt和fileB.txt已保存在本地文件系统/root/step1_files目录中val dataFile = "file:///root/step1_files"val data = sc.textFile(dataFile, 2)/********** Begin **********///第一步：执行过滤操作，把空行丢弃。val rdd1 = data.filter(_.trim().length > 0)//第二步：执行map操作，取出RDD中每个元素，去除尾部空格并生成一个(key, value)键值对。val rdd2 = rdd1.map(line => (line.trim, ""))//第三步：执行groupByKey操作，把所有key相同的value都组织成一个value-list。val rdd3 = rdd2.groupByKey()//第四步：对RDD进行重新分区，变成一个分区，//在分布式环境下只有把所有分区合并成一个分区，才能让所有元素排序后总体有序。val rdd4 = rdd3.partitionBy(new HashPartitioner(1))//第五步：执行sortByKey操作，对RDD中所有元素都按照key的升序排序。val rdd5 = rdd4.sortByKey()//第六步：执行keys操作，将键值对RDD中所有元素的key返回，形成一个新的RDD。val rdd6 = rdd5.keys//第七步：执行collect操作，以数组的形式返回RDD中所有元素。val rdd7 = rdd6.collect()//第八步：执行foreach操作，并使用println打印出数组中每个元素的值。println("") //注意：此行不要修改，否则会影响测试结果，在此行之后继续完成第八步的代码。rdd7.foreach(println)/********** End **********/}
}

spark RDD编程第1关：数据去重相关推荐

spark RDD编程
RDD,也就是弹性分布式数据集的简称,它是spark处理的分布式元素集合. 对于RDD的操作包括: 创建RDD(从外部数据或者内存中的数据),转化RDD(利用筛选条件等),调用RDD操作求值. * ...
Spark RDD编程模型及算子介绍（二）
文章目录常见的Action算子常见分区操作算子常见的Action算子 countByKey算子:统计Key出现的次数,部分代码如下: rdd_file = sc.textFile(". ...
Spark RDD编程API
1. RDD的算子分类 Transformation(转换):根据数据集创建一个新的数据集,计算后返回一个新RDD:例如:一个rdd进行map操作后生了一个新的rdd. Action(动作):对rdd ...
spark应用程序转换_Spark—RDD编程常用转换算子代码实例
Spark-RDD编程常用转换算子代码实例 Spark rdd 常用 Transformation 实例: 1.def map[U: ClassTag](f: T => U): RDD[U] ...
Spark RDD（三）
文章目录 1. 前言 2. 数据源 3. 调度器 4. RDD 编程接口 4.1 数据源自于集合的接口 4.2 数据源自于RDD的接口 4.3 数据源自于DataFrame的接口 5. RDD操作 1 ...
【大数据开发】SparkCore——Spark作业执行流程、RDD编程的两种方式、简单算子
文章目录一.Spark作业执行流程(重点) 二.RDD编程 2.1创建RDD的⼆种⽅式: 2.2Transformation算⼦ 2.3Action算子三.简单算子(必须掌握) 3.1 map.m ...
第二章 Spark RDD以及编程接口
第二章 Spark RDD以及编程接口目录 Spark程序"Hello World" Spark RDD 创建操作转换操作控制操作行动操作注:学习<Spark大数据 ...
Spark综合大作业：RDD编程初级实践
Spark综合大作业:RDD编程初级实践实验配置:操作系统:Ubuntu16.04 | 环境:Spark版本:2.4.0 | 软件:Python版本:3.4.3. 文章目录一.实验目的二.实验平 ...
spark期末大作业RDD编程初级实践
1.需求描述本次实验需要:系统:linux unbuntu14.04,处理器:至少需要两个处器,一个内核,内存:至少4G,硬盘空间:大小需要20GB.Hadoop:2.7.1以上版本,JDK:1.8 ...

spark RDD编程第1关：数据去重

任务描述

RDD的转换操作

RDD的行动操作

编程要求

代码

spark RDD编程第1关：数据去重相关推荐

最新文章

热门文章

spark RDD编程 第1关：数据去重

任务描述

RDD的转换操作

RDD的行动操作

编程要求

代码

spark RDD编程 第1关：数据去重相关推荐

最新文章

热门文章

spark RDD编程第1关：数据去重

spark RDD编程第1关：数据去重相关推荐