RDD是如何操作数据转换的？RDD转换算子API示例

　　RDD处理过程中的“转换”操作主要用于根据已有RDD创建新的RDD，每一次通过Transformation算子计算后都会返回一个新RDD，供给下一个转换算子使用。下面，通过一张表来列举一些常用转换算子操作的API，如表1所示。

　　表1 常用的转换算子API

　　下面，我们通过结合具体的示例对这些转换算子API进行详细讲解。

　　·filter(func)

　　filter(func)操作会筛选出满足函数func的元素，并返回一个新的数据集。假设，有一个文件test.txt(内容如文件3-1)，下面，通过一张图来描述如何通过filter算子操作，筛选出包含单词“spark”的元素，具体过程如图1所示。

　　 图1 filter算子操作

　　在图1中，通过从test.txt文件中加载数据的方式创建RDD，然后通过filter操作筛选出满足条件的元素，这些元素组成的集合是一个新的RDD。接下来，通过代码来进行演示，具体代码如下：

  scala> val lines = sc.textFile("file:///export/data/test.txt")
  lines: org.apache.spark.rdd.RDD[String] = `[file:///export/data/test.txt]`(file:///\\export\data\test.txt)
              MapPartitionsRDD[1] at textFile at :24
  scala> val linesWithSpark = lines.filter(line => line.contains("spark"))
  linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at
                             filter at :25

　　在上述代码中，filter()输入的参数line => line.contains(“spark”)是一个匿名函数，其含义是依次取出lines这个RDD中的每一个元素，对于当前取到的元素，把它赋值给匿名函数中的line变量。若line中包含“spark”单词，就把这个元素加入到RDD(即linesWithSpark)中，否则就丢弃该元素。

　　·map(func)

　　map(func)操作将每个元素传递到函数func中，并将结果返回为一个新的数据集。假设，有一个文件test.txt(内容如文件1)，接下来，通过一张图来描述如何通过map算子操作把文件内容拆分成一个个的单词并封装在数组对象中，具体过程如图2所示。

　图2 map算子操作

　　在图2中，通过从test.txt文件中加载数据的方式创建RDD，然后通过map操作将文件的每一行内容都拆分成一个个的单词元素，这些元素组成的集合是一个新的RDD。接下来，通过代码来进行演示，具体代码如下：

  scala> val lines = sc.textFile("file:///export/data/test.txt")
  lines: org.apache.spark.rdd.RDD[String] = [file:///export/data/test.txt](file:///\\export\data\test.txt)
              MapPartitionsRDD[4] at textFile at :24
  scala> val words = lines.map(line => line.split(" "))
  words: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[13] at
                             map at :25

　　上述代码中，lines.map(line => line.split(“ ”))含义是依次取出lines这个RDD中的每个元素，对于当前取到的元素，把它赋值给匿名函数中的line变量。由于line是一行文本，如“hadoop spark”，一行文本中包含多个单词，且空格进行分隔，通过line.split(“ ”)匿名函数，将文本分成一个个的单词，拆分后得到的单词都被封装到一个数组对象中，成为新的RDD(即words)的一个元素。

　　·flatMap(func)

　　flatMap(func)与map(func)相似，但是每个输入的元素都可以映射到0或者多个输出的结果。有一个文件test.txt(内容如文件3-1)，接下来，通过一张图来描述如何通过flatMap算子操作，把文件内容拆分成一个个的单词，具体过程如图3所示。

图3 flatMap算子操作

　　在图3中，通过从test.txt文件中加载数据的方式创建RDD，然后通过flatMap操作将文件的每一行内容都拆分成一个个的单词元素，这些元素组成的集合是一个新的RDD。接下来，通过代码来进行演示，具体代码如下：

 scala> val lines = sc.textFile("file:///export/data/test.txt")
  lines: org.apache.spark.rdd.RDD[String] = [file:///export/data/test.txt](file:///\\export\data\test.txt)
              MapPartitionsRDD[5] at textFile at :24
  scala> val words = lines.flatMap(line => line.split(" "))
  words: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[14] at
                             map at :25

　　在上述代码中，lines. flatMap(line => line.split(“ ”))等价于先执行lines.map(line => line.split(“ ”))操作(请参考map(func)操作)，再执行flat()操作(即扁平化操作)，把wordArray中的每个RDD都扁平成多个元素，被扁平后得到的元素构成一个新的RDD(即words)。

　　groupByKey()

　　groupByKey()主要用于(Key，Value)键值对的数据集，将具有相同Key的Value进行分组，会返回一个新的(Key，Iterable)形式的数据集。同样以文件test.txt为例，接下来，通过一张图来描述如何通过groupByKey算子操作，将文件内容中的所有单词进行分组，具体过程如图4示。

图4 groupByKey算子操作

　　在图4中，通过groupByKey操作把(Key，Value)键值对类型的RDD，按单词将单词出现的次数进行分组，这些元素组成的集合是一个新的RDD。接下来，通过代码来进行演示，具体代码如下：

scala> val lines = sc.textFile("file:///export/data/test.txt")
  lines: org.apache.spark.rdd.RDD[String] = [file:///export/data/test.txt](file:///\\export\data\test.txt)             MapPartitionsRDD[6] at textFile at :24
  scala> val words=lines.flatMap(line=>line.split(" ")).map(word=>(word,1))
  words: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[15] at                            map at :25
  scala> val groupWords=words.groupByKey()
   groupWords: org.apache.spark.rdd.RDD[(String,Iterable[Int])]=ShuffledRDD[16]
                        at groupByKey at :25

　　上述代码中，words.groupByKey()操作执行后，RDD中所有的Key相同的Value都被合并到一起。例如，(“spark”，1)、(“spark”，1)、(“spark”，1)这三个键值对的Key都是“spark”，合并后得到新的键值对(“spark”，(1，1，1))。

　　reduceByKey(func)

　　reduceByKey()主要用于(Key，Value)键值对的数据集，返回的是一个新的(Key，Iterable)形式的数据集，该数据集是每个Key传递给函数func进行聚合运算后得到的结果。同样以文件test.txt(内容如文件3-1)，接下来，通过一张图来描述如何通过reduceByKey算子操作统计单词出现的次数，具体操作如图5所示。

 图5 reduceByKey()算子操作

　　在图5中，通过reduceByKey操作把(Key，Value)键值对类型的RDD，按单词Key将单词出现的次数Value进行聚合，这些元素组成的集合是一个新的RDD。接下来，通过代码来进行演示，具体代码如下：

 scala> val lines = sc.textFile("file:///export/data/test.txt")  lines: org.apache.spark.rdd.RDD[String] = [file:///export/data/test.txt](file:///\\export\data\test.txt)
              MapPartitionsRDD[7] at textFile at :24
  scala> val words=lines.flatMap(line=>line.split(" ")).map(word=>(word,1))  words: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[16] at                              map at :25
  scala> val reduceWords=words.reduceByKey((a,b)=>a+b)
  reduceWords: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[17] at
                         reduceByKey at :25

　　上述代码中，执行words.reduceByKey((a，b) => a + b)操作，共分为两个步骤，分别是先执行reduceByKey()操作，将所有Key相同的Value值合并到一起，生成一个新的键值对(例如(“spark”，(1，1，1)));然后执行函数func的操作，即使用(a，b)=> a + b函数把(1，1，1)进行聚合求和，得到最终的结果，即(“spark”，3)。    

怎样使用Spark Shell来读取HDFS文件

HBase数据库是怎样存储数据的？

怎样使用Linux和HDFS创建RDD？

传智大数据培训课程