Spark的RDD简单操作

约定不等于承诺〃 2022-07-24 05:18 152阅读 0赞

**0、Spark的wc.note**

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:9pt; background-color:rgb(255,255,255)"><pre><code>
</code></pre><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * hadoop
  * spark
  * tachyon
  * hadoop
  * hbase
  * spark
  */
/**
  * Created by Administrator on 2016/4/23.
  */
object rdd0_wc {
def main(args: Array[String]) {
//创建环境变量
val conf=new SparkConf().setMaster("local").setAppName("rdd0_wc")
//创建环境变量实例
val sc=new SparkContext(conf)
//读取文件
val data=sc.textFile("E://sparkdata//wc.txt")
    data.flatMap(_.split("\t")).map((_,1)).reduceByKey(_+_).collect().foreach(println)  //wrod计数
}
}</code></pre><br><br><br></pre> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center][]

![Image 1][]

**1、rdd1\_aggregate.note**

--------------------

aggregate用法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/23.
  */
object rdd1_aggregate {
/**
    * aggregate方法
    *  RDD是工作在Spark上，因此，parallelize方法是将内存数据读入Spark系统中，作为一个整体的数据集。
    * math.max方法用于比较数据集中数据的大小；
    * 第二个_+_方法是对传递的第一个比较方法结果进行处理。第一个比较结果是6，与中立空值相加，所以最终结果为6
    */
def main(args: Array[String]) {
val conf=new SparkConf().setMaster("local").setAppName("rdd1_aggregate")
val sc=new SparkContext(conf)
val arr=sc.parallelize(Array(1,2,3,4,5,6))
val result=arr.aggregate(0)(math.max(_, _), _ + _)  //aggregate用法
println(result)
  }

/**
    * 参数改变后
    * 这里parallelize将数据分成两个节点存储
    * math方法分别查找出两个数据集的最大值，分别是3和6.
    * 这样在调用aggregate方法的第二个计算方法时，将查找的数据值进行相加，获得最大值9
    */
//    def main(args: Array[String]) {
//      val conf=new SparkConf().setMaster("local").setAppName("rdd1_aggregate")
//      val sc=new SparkContext(conf)
//      val arr=sc.parallelize(Array(1,2,3,4,5,6),2)
//      val result=arr.aggregate(0)(math.max(_, _), _ + _)  //aggregate用法
//      println(result)
//    }

/**
    * aggregate方法用于字符串
    */
//    def main(args: Array[String]) {
//      val conf=new SparkConf().setMaster("local").setAppName("rdd1_aggregate")
//      val sc=new SparkContext(conf)
//      val arr=sc.parallelize(Array("hadoop","spark","hive","hbase","kafka"))
//      val result=arr.aggregate("")((value,word)=&gt;value+"、"+word,_+_)  //aggregate用法
//      println(result)
//  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

示例一结果：

（   **val ** arr=sc.parallelize( *Array* ( 1 , 2 , 3 , 4 , 5 , 6 )) ）

![Image 1][]

![Center 1][]

示例二结果：

（ *val arr=sc.parallelize(Array(1,2,3,4,5,6),2)* ）

![Image 1][] ![Center 2][]

示例三结果：

（ *val arr=sc.parallelize(Array("hadoop","spark","hive","hbase","kafka"))* ）

![Image 1][]

![Center 3][]

**2、rdd2\_cache.note**

--------------------

cache()简单用法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  * cache方法的作用是将数据内容计算并保存在计算节点的内存中，这个方法的使用是针对Spark的Lazy数据处理模式。
  * 在Lazy模式中，数据在编译和使用时是不进行计算的，而仅仅保存其存储地址，只有Action方法到来时才正式计算。
  * 这样做的好处是可以极大的减少存储空间，从而提高利用率，而有时必须要求数据进行计算，此时就需要使用cache方法。
  */
object rdd2_cache {
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd2_cache") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array("hadoop","spark","hive"))  //设定数据集
println(arr)  //打印结果
val arr1=arr.cache()
println("-----------------------------")  //分隔符
arr1.foreach(println)  //专门用来打印未进行Action操作的数据的专用方法，可以对数据进行提早计算。
}
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 4][]

![Image 1][]

**3、rdd3\_cartesian.note**

--------------------

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  * 笛卡尔积操作cartsian方法
  */
object rdd3_cartesian {
/**
    * 此方法用于对不同的数组进行笛卡尔积操作，要求是数据集的长度必须相同，结果作为一个新的数据集返回
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd3_cartesian") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5))  //创建第一个数组
val arr1=sc.parallelize(Array(6,7,8,9,10))  //创建第二个数组
val result=arr.cartesian(arr1)  //进行笛卡尔积计算
result.foreach(println) //打印结果
}
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 5][]

![Image 1][]

**4、rdd4\_coalesce.note**

--------------------

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd4_coalesce {
/**
    * coalesce方法是将已经存储的数据重新分片后再进行存储
    * 第一个参数是将数据重新分成的片数，布尔型数指的是将数据分成更小的片时使用。
    */
//  def main(args: Array[String]) {
//    val conf=new SparkConf()  //创建环境变量
//      .setMaster("local") //设置本地化处理
//      .setAppName("rdd4_coalesce") //设置名称
//    val sc=new SparkContext(conf)   //创建环境变量实例
//    val arr=sc.parallelize(Array(1,2,3,4,5,6))
//    val arr2=arr.coalesce(2,true) //将数据重新分区
//    val result=arr.aggregate(0)(math.max(_,_),_+_)  //计算数据值
//    println(result)
//    val result2=arr2.aggregate(0)(math.max(_,_),_+_)  //计算重新分区数据值
//    println(result2)
//  }

/**
    * RDD中还有一个repartition方法与这个coalesce方法类似，均是将数据重新分区组合
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd4_coalesce") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5,6))
val arr2=arr.repartition(3) //将数据分区
println(arr2.partitions.length) //打印分区结果
}
}</code></pre><br><span style="background-color:inherit; color:rgb(0,0,128)"><strong>package </strong></span>RddApi<br style="background-color:inherit"><br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>import </strong></span>org.apache.spark.{SparkConf, SparkContext}<br style="background-color:inherit"><br style="background-color:inherit"><span style="background-color:inherit; color:rgb(255,0,0)"><em>/**<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * Created by Administrator on 2016/4/24.<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> */<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>object </strong></span>rdd4_coalesce {
          <br style="background-color:inherit"><span style="background-color:inherit; color:rgb(255,0,0)"><em>/**<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * coalesce方法是将已经存储的数据重新分片后再进行存储<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * 第一个参数是将数据重新分成的片数，布尔型数指的是将数据分成更小的片时使用。<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> */<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// def main(args: Array[String]) { <br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val conf=new SparkConf() //创建环境变量<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// .setMaster("local") //设置本地化处理<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// .setAppName("rdd4_coalesce") //设置名称<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val sc=new SparkContext(conf) //创建环境变量实例<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val arr=sc.parallelize(Array(1,2,3,4,5,6))<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val arr2=arr.coalesce(2,true) //将数据重新分区<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val result=arr.aggregate(0)(math.max(_,_),_+_) //计算数据值<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// println(result)<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// val result2=arr2.aggregate(0)(math.max(_,_),_+_) //计算重新分区数据值<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// println(result2)<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>// }<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em><br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> /**<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * RDD中还有一个repartition方法与这个coalesce方法类似，均是将数据重新分区组合<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> */<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>def </strong></span>main(args: Array[<span style="background-color:inherit; color:rgb(32,153,157)">String</span>]) {
          <br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>conf=<span style="background-color:inherit; color:rgb(0,0,128)"><strong>new </strong></span>SparkConf()  <span style="background-color:inherit; color:rgb(255,0,0)"><em>//创建环境变量<br style="background-color:inherit"></em></span>.setMaster(<span style="background-color:inherit; color:rgb(0,128,0)"><strong>"local"</strong></span>) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//设置本地化处理<br style="background-color:inherit"></em></span>.setAppName(<span style="background-color:inherit; color:rgb(0,128,0)"><strong>"rdd4_coalesce"</strong></span>) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//设置名称<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>sc=<span style="background-color:inherit; color:rgb(0,0,128)"><strong>new </strong></span>SparkContext(conf)   <span style="background-color:inherit; color:rgb(255,0,0)"><em>//创建环境变量实例<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>arr=sc.parallelize(<span style="background-color:inherit"><em>Array</em></span>(<span style="background-color:inherit; color:rgb(0,0,255)">1</span>,<span style="background-color:inherit; color:rgb(0,0,255)">2</span>,<span style="background-color:inherit; color:rgb(0,0,255)">3</span>,<span style="background-color:inherit; color:rgb(0,0,255)">4</span>,<span style="background-color:inherit; color:rgb(0,0,255)">5</span>,<span style="background-color:inherit; color:rgb(0,0,255)">6</span>))<br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>arr2=arr.repartition(<span style="background-color:inherit; color:rgb(0,0,255)">3</span>) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//将数据分区<br style="background-color:inherit"></em></span><span style="background-color:inherit"><em>println</em></span>(arr2.partitions.length) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//打印分区结果<br style="background-color:inherit"></em></span>}<br style="background-color:inherit">}<br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

示例一：

![Image 1][]

![Center 6][]

![Image 1][]

![Center 7][]

示例2：

![Image 1][]

![Center 8][]

**5、rdd5\_countByValue.note**

--------------------

countByValue用法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd5_countByValue {
/**
    * countByValue方法是计算数据集中某个数据出现的个数，并将其以map的形式返回
    * @param args
*/
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd5_countByValue") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,1,2,2,3,3,4,5,6))
val result=arr.countByValue() //计算个数
result.foreach(print)
  }
}</code></pre><br><span style="background-color:inherit; color:rgb(0,0,128)"><strong>package </strong></span>RddApi<br style="background-color:inherit"><br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>import </strong></span>org.apache.spark.{SparkConf, SparkContext}<br style="background-color:inherit"><br style="background-color:inherit"><span style="background-color:inherit; color:rgb(255,0,0)"><em>/**<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * Created by Administrator on 2016/4/24.<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> */<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>object </strong></span>rdd5_countByValue {
           <br style="background-color:inherit"><span style="background-color:inherit; color:rgb(255,0,0)"><em>/**<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * countByValue方法是计算数据集中某个数据出现的个数，并将其以map的形式返回<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(255,0,0)"><em> * </em></span><span style="background-color:inherit"><strong>@param args<br style="background-color:inherit"></strong></span><span style="background-color:inherit; color:rgb(255,0,0)"><em>*/<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>def </strong></span>main(args: Array[<span style="background-color:inherit; color:rgb(32,153,157)">String</span>]) {
           <br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>conf=<span style="background-color:inherit; color:rgb(0,0,128)"><strong>new </strong></span>SparkConf()  <span style="background-color:inherit; color:rgb(255,0,0)"><em>//创建环境变量<br style="background-color:inherit"></em></span>.setMaster(<span style="background-color:inherit; color:rgb(0,128,0)"><strong>"local"</strong></span>) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//设置本地化处理<br style="background-color:inherit"></em></span>.setAppName(<span style="background-color:inherit; color:rgb(0,128,0)"><strong>"rdd5_countByValue"</strong></span>) <span style="background-color:inherit; color:rgb(255,0,0)"><em>//设置名称<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>sc=<span style="background-color:inherit; color:rgb(0,0,128)"><strong>new </strong></span>SparkContext(conf)   <span style="background-color:inherit; color:rgb(255,0,0)"><em>//创建环境变量实例<br style="background-color:inherit"></em></span><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>arr=sc.parallelize(<span style="background-color:inherit"><em>Array</em></span>(<span style="background-color:inherit; color:rgb(0,0,255)">1</span>,<span style="background-color:inherit; color:rgb(0,0,255)">1</span>,<span style="background-color:inherit; color:rgb(0,0,255)">2</span>,<span style="background-color:inherit; color:rgb(0,0,255)">2</span>,<span style="background-color:inherit; color:rgb(0,0,255)">3</span>,<span style="background-color:inherit; color:rgb(0,0,255)">3</span>,<span style="background-color:inherit; color:rgb(0,0,255)">4</span>,<span style="background-color:inherit; color:rgb(0,0,255)">5</span>,<span style="background-color:inherit; color:rgb(0,0,255)">6</span>))<br style="background-color:inherit"><span style="background-color:inherit; color:rgb(0,0,128)"><strong>val </strong></span>result=arr.countByValue() <span style="background-color:inherit; color:rgb(255,0,0)"><em>//计算个数<br style="background-color:inherit"></em></span>result.foreach(<span style="background-color:inherit"><em>print</em></span>)<br style="background-color:inherit">  }<br style="background-color:inherit">}<br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 9][]

![Image 1][]

**6、rdd6\_countByKey.note**

--------------------

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd6_countByKey {
/**
    * countByKey方法与countByValue方法有本质的区别。
    * countByKey是计算数组中元数据键值对Key出现的个数
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd6_countByKey") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array((1,"hadoop"),(2,"spark"),(3,"tachyon")))
val result=arr.countByKey() //进行计数
result.foreach(print)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 10][]

![Image 1][]

**7、rdd7\_distinct.note**

--------------------

distinct方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd7_distinct {
/**
    * distinct方法作用是去除数据集中重复项
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd7_distinct") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(("hadoop"),("spark"),("hive"),("hadoop"),("hive")))
val result=arr.distinct()
    result.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 11][]

![Image 1][]

**8、rdd8\_filter.note**

--------------------

filter方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd8_filter {
/**
    * filter方法用来对数据集进行过滤
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd8_filter") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5))
val result=arr.filter(_&gt;=3)
    result.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> <br style="background-color:inherit"> </td> 
  </tr> 
 </tbody> 
</table>

![Center 12][]

![Image 1][]

**9、rdd9\_flatMap.note**

--------------------

flatMap方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd9_flatMap {
def main(args: Array[String]) {
/**
      * flatMap方法是对RDD中的数据集进行整体操作的一个特殊方法。
      */
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd9_flatMap") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5,6,7))
val result=arr.flatMap(x=&gt;List(x+1)).collect()  //进行数据集计算
result.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 13][]

![Image 1][]

**10、rdd10\_map .note**

--------------------

map方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd10_map {
/**
    * map方法是对RDD中的数据集中的数据进行逐个处理。
    * map与flatMap不同之处在于，flatMap是将数据集中的数据作为一个整体去处理，之后再对其中的数据做计算。
    * 而map方法直接对数据集中的数据做单独的处理。
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd10_map") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5,6))
val result=arr.map(x=&gt;List(x+1)).collect()
    result.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 14][]

![Image 1][]

**11、rdd11\_groupBy .note**

--------------------

group方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd11_groupBy {
/**
    * groupBy方法是将传入的数据进行分组
    * 传入的第一个参数是方法名，第二个参数是分组的标签
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd11_groupBy") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array(1,2,3,4,5,6))
    arr.groupBy(myFilter1).foreach(println)  //设置第一个分组
}
def myFilter1(num:Int):Int={ //自定义第一个分组
num %2
}

}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 15][]

![Image 1][]

**12、rdd12\_keyBy .note**

--------------------

keyBy方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd12_keyBy {
/**
    * keyBy方法是为数据集中的每个个体数据增加一个Key，从而可以和原来的数据集形成键值对。
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd12_keyBy") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr=sc.parallelize(Array("hadoop","spark","tachyon","hive","hbase"))
val str=arr.keyBy(word=&gt;word.size)  //设置配置方法
str.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 16][]

![Image 1][]

**13、rdd13\_reduce.note**

--------------------

reduce方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd13_reduce {
/**
    * reduce方法主要是对传入的数据进行合并处理。
    * 第一个下划线代表数据集中的第一个参数。
    * 第二个下划线在第一次合并处理时代表空集
    */
def main(args: Array[String]) {
//    val conf=new SparkConf()  //创建环境变量
//      .setMaster("local") //设置本地化处理
//      .setAppName("rdd13_reduce") //设置名称
//    val sc=new SparkContext(conf)   //创建环境变量实例
//    val arr=sc.parallelize(Array("hadoop","spark","tachyon","hive","hbase"))
//    val result=arr.reduce(_+_)
//    result.foreach(print)

/**
      * 寻找最长字符串
      */
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd13_reduce") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val str=sc.parallelize(Array("hadoop","spark","tachyon","hive"))
val result=str.reduce(myFun)  //进行数据拟合
result.foreach(print)
  }
def myFun(str1:String,str2:String):String={
var str=str1
if(str2.size&gt;=str.size){
      str=str2
    }
return str
  }
}
</code></pre><br><br></pre> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

示例一：

![Image 1][]

![Center 17][]

示例二：

![Image 1][]

![Center 18][]

**14、rdd14\_sortBy.note**

--------------------

sortBy方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  */
object rdd14_sortBy {
/**
    * sortBy方法是一个常用的排序方法，其功能是对已有的RDD重新排序，并将重新排序后的数据生成一个新的RDD。
    * sortBy有3个参数：
    * 第一个参数：传入方法，用以计算排序的方法
    * 第二个参数：指定排序的值按升序还是降序显示
    * 第三个参数：分片的数量
    */
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd14_sortBy") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val str=sc.parallelize(Array((1,"hadoop"),(2,"spark"),(3,"hive"),(4,"hbase"),(5,"tachyon")))
val str1=str.sortBy(word=&gt;word._1,true)  //按照第一个数据排序
val str2=str.sortBy(word=&gt;word._2,true) //按照第二个数据排序
str1.foreach(println)
    str2.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

按照第一个数据排序：

![Image 1][] ![Center 19][]

按照第二个数据排序：

![Image 1][] ![Center 20][]

**15、rdd15\_zip.note**

--------------------

zip方法：

<table style="font-size:undefined; border-collapse:collapse; margin-top:10px; margin-bottom:10px; background-color:inherit"> 
 <tbody style="background-color:inherit"> 
  <tr style="background-color:inherit"> 
   <td style="word-break:break-all; border:1px solid rgb(153,153,153); padding:5px 16px 5px 12px; min-height:25px; min-width:25px; height:25px; background-color:inherit; width:579px"> 
    <div style="min-width:2px; background-color:inherit"> 
     <pre style="font-family:宋体; font-size:10.5pt; background-color:rgb(255,255,255)"><pre><code>package RddApi

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2016/4/24.
  * zip方法是常用的合并压缩算法，它将若干个RDD压缩成一个新的DD，进而形成一系列的键值对存储形式的新RDD。
  */
object rdd15_zip {
def main(args: Array[String]) {
val conf=new SparkConf()  //创建环境变量
.setMaster("local") //设置本地化处理
.setAppName("rdd15_zip") //设置名称
val sc=new SparkContext(conf)   //创建环境变量实例
val arr1=sc.parallelize(Array(1,2,3,4,5,6))
val arr2=sc.parallelize(Array("a","b","c","d","e","f"))
val arr3=sc.parallelize(Array("g","h","i","j","k","l"))
val arr4=arr1.zip(arr2).zip(arr3) //进行压缩算法
arr4.foreach(println)
  }
}</code></pre><br><br style="background-color:inherit"></pre> 
     <br style="background-color:inherit"> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

![Center 21][]

![Image 1][]

[Center]: /images/20220722/7b63371c055342b2b5209617297f042d.png
[Image 1]: 
[Center 1]: /images/20220722/c642fe2c60b34cfba6b817118eed242e.png
[Center 2]: /images/20220722/d80190e86cee474a91d5f0401196e477.png
[Center 3]: /images/20220722/f4ea3edccd4e47a19f5c1d0ecad7a09b.png
[Center 4]: /images/20220722/b80aecbd95c84d31a3b109b4948db6b8.png
[Center 5]: /images/20220722/afd3622f8e06452497ff0c9d8c4768c2.png
[Center 6]: /images/20220722/bb8261c432514d40a59c6bab2ff306b9.png
[Center 7]: /images/20220722/eb194498d4b14ceb8b4da03636ac5a94.png
[Center 8]: /images/20220722/fc6711b9f3b946d8b88055b1fe3eb2eb.png
[Center 9]: /images/20220722/e3b47e113f314745b971ebe52970d785.png
[Center 10]: /images/20220722/bdcefed547364f72a4f9bbd404e557f2.png
[Center 11]: /images/20220722/926fe7d9f0154dd091f99951d92e9f12.png
[Center 12]: /images/20220722/a1c78a4865134a88b09312770b7e0c6d.png
[Center 13]: /images/20220722/f3a176fdb19548b585e987e9d9bdd34e.png
[Center 14]: /images/20220722/93d3bce38fea46f7ba65052fcf5d36e3.png
[Center 15]: /images/20220722/60e83bbe55244c138a089a11ac714765.png
[Center 16]: /images/20220722/d8c8dca423504ed3b5f2c7e5345e4de8.png
[Center 17]: /images/20220722/3ee22a9f5fc44c2bb9ddf85fd524b93b.png
[Center 18]: /images/20220722/6b18fa392882439387fd16f9a2b927d8.png
[Center 19]: /images/20220722/b967db4f1ba6481a800174dec84226e9.png
[Center 20]: /images/20220722/a51dac5e69424db3981b4dfe7128255e.png
[Center 21]: /images/20220722/fc8c90dc56fc4b1f8e15d51eced44cea.png