RDD案例（DT大数据梦工厂）

标签：rdd案例

内容：

1、map、filter、flatmap等操作回顾；

2、reduceBykey、groupBykey；

3、jion、cogroug；

算子共同特点：都是最常用的算子，构建复杂算法的基石，都是lazy级别的，不属于action

创建SparkContext是Spark的起点，只有创建SparkContext，才能创建RDD

==========map============

适用于任何元素且对其作用的集合中的每一个元素循环遍历，并调用其作为参数的函数对每一个遍历的元素进行具体化处理

==========filter============

符合条件的留下，不符合条件的去除

==========flatmap============

首先通过传入的参数的函数来作用于RDD的每个字符串进行单词切分，然后把切分后的结果合并成一个大的集合(是以集合的方式存在的)

package com.dt.spark.cores

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

/**
  * 最常用，最重要的Spark Transformation案例实战
  * Created by 威 on 2016/2/8.
  */
object Transformations {
  def main(args: Array[String]) {
    val conf = new SparkConf()//创建SparkConf对象
    conf.setAppName("Transformations")//设置应用程序的名称，在程序运行的监控界面可以看到名称
    conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
    val sc = new SparkContext(conf)//通过创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息，这是第一个RDD创建的唯一入口，也是Drvier灵魂，是通往集群的唯一通道

    /**
      * Map操作
      *总结：适用于任何元素且对其作用的集合中的每一个元素循环遍历，并调用其作为参数的函数对每一个遍历的元素进行具体化处理
      */
    val nums = sc.parallelize(1 to 10)//根据集合创建RDD
    val mapped = nums.map(item=>2*item)//map就是循环遍历集合中的每个元素，进行函数操作，map知道是整数，是Scala进行了类型推断
    mapped.collect().foreach(println)

    /**
      * Filter
      */
    val filtered = nums.filter(item=>item%2==0)
    filtered.collect().foreach(println)

    /**
      * Flatmap
      */
    val bigData = Array("Scala Spark","Java Hadoop","Java Tachyon")
    val bigDataStrings = sc.parallelize(bigData)
    val words = bigDataStrings.flatMap(line => line.split(" "))
    words.collect().foreach(println)

    /**
      * 上面代码最大的问题：不够简洁，应该函数式编程
      */

    sc.stop()

  }
}

改进：

package com.dt.spark.cores

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

/**
  * Created by 威 on 2016/2/8.
  */
object Transformations_advanced {
  def main(args: Array[String]) {
    val sc = sparkContext("Transformations")//创建SparkContext
    mapTransformation(sc)//map案例
    filterTransformation(sc)//filter案例
    flatMapTransformation(sc)//flatMap案例
    sc.stop()//停止SparkContext，销毁相关的Drvier对象，释放资源
  }

  def sparkContext(name:String) ={
    val conf = new SparkConf()//创建SparkConf对象
    conf.setAppName(name)//设置应用程序的名称，在程序运行的监控界面可以看到名称
    conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
    val sc = new SparkContext(conf)//通过创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息，这是第一个RDD创建的唯一入口，也是Drvier灵
    sc
  }

  def mapTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 10)//根据集合创建RDD
    val mapped = nums.map(item=>2*item)//map就是循环遍历集合中的每个元素，进行函数操作，map知道是整数，是Scala进行了类型推断
    mapped.collect().foreach(println)//搜集运行结果并通过foreach打印
  }

  def filterTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 20)//根据集合创建RDD
    val filtered = nums.filter(item=>item%2==0)//根据参数中函数的Bool值来判断符合条件的元素，并基于这些元素构成新的maoPatitionsRDD
    filtered.collect().foreach(println)
  }

  def flatMapTransformation(sc:SparkContext): Unit ={
    val bigData = Array("Scala Spark","Java Hadoop","Java Tachyon")//实例化字符串类型的Array
    val bigDataStrings = sc.parallelize(bigData)//创建以字符串元素为类型的ParallelCollectRDD
    val words = bigDataStrings.flatMap(line => line.split(" "))//首先通过传入的参数的函数来作用于RDD的每个字符串进行单词切分，然后把切分后的结果合并成一个大的集合(是以集合的方式存在的)，产生结果为Scala Spark Java Hadoop Java Tachyon
    words.collect().foreach(println)
  }
}

总结：main方法里面调用的每一个功能都必须是模块化的，每个模块可以使用函数封装

==========groupByKey============

按照相同的value进行分组，分组后的value是一个集合

==========reduceByKey============

wordcount已经讲过

代码：

package com.dt.spark.cores

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

/**
  * Created by 威 on 2016/2/8.
  */
object Transformations_advanced {
  def main(args: Array[String]) {
    val sc = sparkContext("Transformations")//创建SparkContext
    mapTransformation(sc)//map案例
    filterTransformation(sc)//filter案例
    flatMapTransformation(sc)//flatMap案例
    groupByKeyTransformation(sc)//groupByKey案例
    reduceByKeyTransformation(sc)//reduceByKey案例

    sc.stop()//停止SparkContext，销毁相关的Drvier对象，释放资源
  }

  def sparkContext(name:String) ={
    val conf = new SparkConf()//创建SparkConf对象
    conf.setAppName(name)//设置应用程序的名称，在程序运行的监控界面可以看到名称
    conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
    val sc = new SparkContext(conf)//通过创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息，这是第一个RDD创建的唯一入口，也是Drvier灵
    sc
  }

  def mapTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 10)//根据集合创建RDD
    val mapped = nums.map(item=>2*item)//map就是循环遍历集合中的每个元素，进行函数操作，map知道是整数，是Scala进行了类型推断
    mapped.collect().foreach(println)//搜集运行结果并通过foreach打印
  }

  def filterTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 20)//根据集合创建RDD
    val filtered = nums.filter(item=>item%2==0)//根据参数中函数的Bool值来判断符合条件的元素，并基于这些元素构成新的maoPatitionsRDD
    filtered.collect().foreach(println)
  }

  def flatMapTransformation(sc:SparkContext): Unit ={
    val bigData = Array("Scala Spark","Java Hadoop","Java Tachyon")//实例化字符串类型的Array
    val bigDataStrings = sc.parallelize(bigData)//创建以字符串元素为类型的ParallelCollectRDD
    val words = bigDataStrings.flatMap(line => line.split(" "))//首先通过传入的参数的函数来作用于RDD的每个字符串进行单词切分，然后把切分后的结果合并成一个大的集合(是以集合的方式存在的)，产生结果为Scala Spark Java Hadoop Java Tachyon
    words.collect().foreach(println)
  }

  def groupByKeyTransformation(sc:SparkContext): Unit ={
    val data = Array(Tuple2(100,"Spark"),Tuple2(100,"Tachyon"),Tuple2(70,"Hadoop"),Tuple2(80,"Kafka"),Tuple2(80,"HBase"))//准备数据
    val dataRDD = sc.parallelize(data)//创建RDD
    val gbts = dataRDD.groupByKey()//按照相同的value进行分组，分组后的value是一个集合
    for (elem <- gbts.collect()) {
      /**
        *结果：
        * 100:CompactBuffer(Spark, Tachyon)
        * 80:CompactBuffer(Kafka, HBase)
        * 70:CompactBuffer(Hadoop)
        */
      println(elem._1+":"+elem._2)
    }
  }

  def reduceByKeyTransformation(sc:SparkContext): Unit ={
    val lines = sc.textFile("F:/安装文件/操作系统/spark-1.6.0-bin-hadoop2.6/README.md", 1)
    val words = lines.flatMap { line => line.split(" ") }//对每一行的字符串进行单次拆分并把所有行的拆分结果通过flat合并成一个大的单次集合val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)
    val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)
    val wordCounts = pairs.reduceByKey(_+_)//对相同的key，进行value的累加（包括local和Reducer级别同时reduce）
    wordCounts.foreach(wordNumberPair=>println(wordNumberPair._1+":"+wordNumberPair._2))
  }
}

==========jion和cogroup============

jion和cogroup是所有Spark学习者必须掌握的内容，没有任何商量的余地

jion两个实体之间根据key进行连接，是最重要的Spark算子，大数据角度讲，它可以打通数据的孤岛

package com.dt.spark.cores

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

/**
  * Created by 威 on 2016/2/8.
  */
object Transformations_advanced {
  def main(args: Array[String]) {
    val sc = sparkContext("Transformations")//创建SparkContext
    mapTransformation(sc)//map案例
    filterTransformation(sc)//filter案例
    flatMapTransformation(sc)//flatMap案例
    groupByKeyTransformation(sc)//groupByKey案例
    reduceByKeyTransformation(sc)//reduceByKey案例
    joinTransformation(sc)//join案例
    sc.stop()//停止SparkContext，销毁相关的Drvier对象，释放资源
  }

  def sparkContext(name:String) ={
    val conf = new SparkConf()//创建SparkConf对象
    conf.setAppName(name)//设置应用程序的名称，在程序运行的监控界面可以看到名称
    conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
    val sc = new SparkContext(conf)//通过创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息，这是第一个RDD创建的唯一入口，也是Drvier灵
    sc
  }

  def mapTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 10)//根据集合创建RDD
    val mapped = nums.map(item=>2*item)//map就是循环遍历集合中的每个元素，进行函数操作，map知道是整数，是Scala进行了类型推断
    mapped.collect().foreach(println)//搜集运行结果并通过foreach打印
  }

  def filterTransformation(sc:SparkContext): Unit ={
    val nums = sc.parallelize(1 to 20)//根据集合创建RDD
    val filtered = nums.filter(item=>item%2==0)//根据参数中函数的Bool值来判断符合条件的元素，并基于这些元素构成新的maoPatitionsRDD
    filtered.collect().foreach(println)
  }

  def flatMapTransformation(sc:SparkContext): Unit ={
    val bigData = Array("Scala Spark","Java Hadoop","Java Tachyon")//实例化字符串类型的Array
    val bigDataStrings = sc.parallelize(bigData)//创建以字符串元素为类型的ParallelCollectRDD
    val words = bigDataStrings.flatMap(line => line.split(" "))//首先通过传入的参数的函数来作用于RDD的每个字符串进行单词切分，然后把切分后的结果合并成一个大的集合(是以集合的方式存在的)，产生结果为Scala Spark Java Hadoop Java Tachyon
    words.collect().foreach(println)
  }

  def groupByKeyTransformation(sc:SparkContext): Unit ={
    val data = Array(Tuple2(100,"Spark"),Tuple2(100,"Tachyon"),Tuple2(70,"Hadoop"),Tuple2(80,"Kafka"),Tuple2(80,"HBase"))//准备数据
    val dataRDD = sc.parallelize(data)//创建RDD
    val gbts = dataRDD.groupByKey()//按照相同的value进行分组，分组后的value是一个集合
    for (elem <- gbts.collect()) {
      /**
        *结果：
        * 100:CompactBuffer(Spark, Tachyon)
        * 80:CompactBuffer(Kafka, HBase)
        * 70:CompactBuffer(Hadoop)
        */
      println(elem._1+":"+elem._2)
    }
  }

  def reduceByKeyTransformation(sc:SparkContext): Unit ={
    val lines = sc.textFile("F:/安装文件/操作系统/spark-1.6.0-bin-hadoop2.6/README.md", 1)
    val words = lines.flatMap { line => line.split(" ") }//对每一行的字符串进行单次拆分并把所有行的拆分结果通过flat合并成一个大的单次集合val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)
    val pairs = words.map { word => (word,1) }//其实编程了Tuple  (word,1)
    val wordCounts = pairs.reduceByKey(_+_)//对相同的key，进行value的累加（包括local和Reducer级别同时reduce）
    wordCounts.foreach(wordNumberPair=>println(wordNumberPair._1+":"+wordNumberPair._2))
  }

  def joinTransformation(sc:SparkContext): Unit ={
    val studentNames = Array(Tuple2(1,"Spark"),Tuple2(2,"Tachyon"),Tuple2(3,"Hadoop"))
    val studentScores = Array(Tuple2(1,100),Tuple2(2,95),Tuple2(3,65))
    val names = sc.parallelize(studentNames)
    val scores = sc.parallelize(studentScores)
    val studentNameAndScore = names.join(scores)
    /**
      *结果：
      *(1,(Spark,100))
      *(3,(Hadoop,65))
      *(2,(Tachyon,95))
      */

    studentNameAndScore.collect().foreach(println)

  }
}

cogroup认为的大数据中第二重要的算子，相当于key jion所有value，value里面所有的内容都放在iterator里面

package com.dt.spark.SparkApps.cores;

import java.util.Arrays;

import java.util.List;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

public class CogroupOps {

        public static void main(String[] args) {

               // TODO Auto-generated method stub

               /**

               * 1、创建Spark配置对象SparkConf，设置Spark程序的运行时的程序配置信息

               * 例如：通过setMaster来设置程序要连接的Spark集群的 url，

               * 如果设置为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差(例如只有1G的内存)的初学者

               */

              SparkConf conf = new SparkConf().setAppName("Cogroup Transformation").setMaster("local");

               /**

               * 2、创建SparkContext对象

               * SparkContext是Spark程序所有功能的唯一入口，无论采用 Scala、Java、Python、R等都必须要(

               * 不同的语言具体的类名称不同，如果是JAVA，则为JavaSparkContext)

               * SparkContext核心作用：初始化Spark应用程序所运行所需要的核心组件，包括DAGScheduler、TaskScheduler

               * 、SchedulerBackend 同时还会负责Spark程序往Master注册程序等

               * SparkContext是整个Spark应用程序中最为至关重要的一个对象

               */

              JavaSparkContext sc = new JavaSparkContext(conf);// 其地产曾实际上就是Scala的SparkContext

              List<Tuple2<Integer, String>> namesList = Arrays.asList (new Tuple2<Integer, String>(1, "Spark" ),

                            new Tuple2<Integer, String>(2, "Tachyon"), new Tuple2<Integer, String>(3, "Hadoop" ));

              List<Tuple2<Integer, Integer>> scoresList = Arrays.asList (new Tuple2<Integer, Integer>(1, 100),

                            new Tuple2<Integer, Integer>(2, 90), new Tuple2<Integer, Integer>(3, 70),

                            new Tuple2<Integer, Integer>(1, 110), new Tuple2<Integer, Integer>(2, 95),

                            new Tuple2<Integer, Integer>(3, 60));

              JavaPairRDD<Integer, String> names = sc.parallelizePairs(namesList );//java没有类型推导

              JavaPairRDD<Integer, Integer> scroes = sc.parallelizePairs(scoresList );

              JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> nameScores = names .cogroup(scroes);

               nameScores.foreach(new VoidFunction<Tuple2<Integer,Tuple2<Iterable<String>,Iterable<Integer>>>>() {

                      /**

                      *

                      */

                      private static final long serialVersionUID = 1L;

                      @Override

                      public void call(Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> t ) throws Exception {

                            // TODO Auto-generated method stub

                           System. out.println("Student ID:" +t ._1 );

                           System. out.println("Student Name:" +t ._2 ._1 );

                           System. out.println("Student Score:" +t ._2 ._2 );

                           System. out.println("------------" );

                     }

              });

       }

}

结果：

16/02/08 08:56:32 INFO ShuffleBlockFetcherIterator: Started 0 remote fetches in 0 ms

Student ID:1

Student Name:[Spark]

Student Score:[100, 110]

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

Student ID:3

Student Name:[Hadoop]

Student Score:[70, 60]

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

Student ID:2

Student Name:[Tachyon]

Student Score:[90, 95]

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

16/02/08 08:56:33 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 1165 bytes result sent to driver

就是每个学生的ID下面有名字对应的它所有的分数

作业：Scala完成cogroup。cogroup一共有9个应用场景。先写一个，其它的以理解为主。

package com.dt.spark.cores

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

/**
  * Created by 威 on 2016/2/8.
  */
object CoGroupDemo {
  def main(args: Array[String]) {
    val sc = sparkContext("CoGroupDemo");

    val studentNames = Array(Tuple2(1,"Spark"),Tuple2(2,"Tachyon"),Tuple2(3,"Hadoop"))
    val studentScores = Array(Tuple2(1,100),Tuple2(2,95),Tuple2(3,65),Tuple2(1,95),Tuple2(2,90),Tuple2(3,70))

    val names = sc.parallelize(studentNames)
    val scores = sc.parallelize(studentScores)

    val nameScores = names.cogroup(scores)

    nameScores.foreach{item=>println("Student ID:"+item._1);println("Student Name:"+item._2._1);println("Student Score:"+item._2._2);println("------------")}
  }
  def sparkContext(name:String) ={
    val conf = new SparkConf()//创建SparkConf对象
    conf.setAppName(name)//设置应用程序的名称，在程序运行的监控界面可以看到名称
    conf.setMaster("local")//此时程序在本地运行，不需要安装Spark集群
    val sc = new SparkContext(conf)//通过创建SparkContext对象，通过传入SparkConf实例来定制Spark运行的具体参数和配置信息，这是第一个RDD创建的唯一入口，也是Drvier灵
    sc
  }
}

结果：

16/02/08 09:29:56 INFO ShuffleBlockFetcherIterator: Started 0 remote fetches in 1 ms

Student ID:1

Student Name:CompactBuffer(Spark)

Student Score:CompactBuffer(100, 95)

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

Student ID:3

Student Name:CompactBuffer(Hadoop)

Student Score:CompactBuffer(65, 70)

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

Student ID:2

Student Name:CompactBuffer(Tachyon)

Student Score:CompactBuffer(95, 90)

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

16/02/08 09:29:56 INFO Executor: Finished task 0.0 in stage 2.0 (TID 2). 1165 bytes result sent to driver

本文出自 “一枝花傲寒” 博客，谢绝转载！

RDD案例（DT大数据梦工厂）

标签：rdd案例

原文地址：http://feiweihy.blog.51cto.com/6389397/1741394