黑哥讲解spark-RDD编程

朱雀 2022-03-10 06:25 205阅读 0赞

# spark-RDD编程 #

作者：黑哥

一、编程模型

在spark中，RDD被表示为对象，我们通过一系列的 transformations 定义 RDD 之后，再调用 actions触发 RDD 的计算。只需要记住一句话：在 Spark 中，只有遇到 action，才会执行 RDD 的计算(即延迟计算)，action可以是向应用程序返回结果(count, collect 等)，或者是向存储系统保存数据saveAsTextFile

二、RDD的创建

在spark中，创建RDD有三种：（1）从集合中创建RDD；(2)从外部存储创建RDD；（3）从其他RDD创建。

1）从集合中创建RDD：由一个已经存在的 Scala 集合创建，集合并行化

//通过parallelize
    scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8)) 
    //通过makeRDD
    scala> val rdd2 = sc.makeRDD(List(1,2,3))

2）由外部存储系统的数据集创建，包括本地的文件系统，还有所有 Hadoop支持的数据集，比如 HDFS、Cassandra、HBase 等

scala> val rdd3 = sc.textFile("hdfs://hadoop01:9000/wc")

三、 RDD 一般分为数值 RDD 和键值对 RDD

四、Transformation

RDD 中的所有转换都是延迟加载的，也就是说，它们并不会直接计算结果。相反的，它们只是记住这些应用到基础数据集（例如一个文件）上的转换动作。只有当发生一个要求返回结果给 Driver 的动作时，这些转换才会真正运行。这种设计让 Spark 更加有效率地运行。

练习一：

val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
    //对rdd1里的每一个元素乘2然后排序
    val rdd2 = rdd1.map(_ * 2).sortBy(x => x, true)
    //过滤出大于等于十的元素
    val rdd3 = rdd2.filter(_ >= 10)
    //将元素以数组的方式在客户端显示
    rdd3.collect
    scala>val rdd = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10)).map(_ * 2).sortBy(x => x, true)
    scala>rdd.collect	//查看结果
    Array[Int] = Array(2,4,6,8,10,12,14,16,18,20)

练习二：

val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j"))
    //将rdd1里面的每一个元素先切分在压平
    val rdd2 = rdd1.flatMap(_.split(' '))
    rdd2.collect
    scala>val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j")).flatMap(_.split(' '))
    scala>rdd1.collect	//查看结果
    Array[Int] = Array（a,b,c,d,e,f,h,i,j）

练习三：

scala>val rdd2 = sc.parallelize(List(List("a b c","a b b"),List("f b c","a b b"),List("a f c","a h j")))
    //想得到一个一个单词
    scala>rdd2.flatMap(-.flatMap(_.split(" ")))
    scala>rdd2.collect	//查看结果

练习四：

scala>val rdd3 = sc.parallelize(List(1,2,3,4,5))
    scala>val rdd4 = sc.parallelize(List(5，6,7,8,9))

//求并集

scala>rdd3.union(rdd4)	//会生成新的RDD
    scala>rdd5.collect	//查询结果
    rdd6:Array[Int] = Array(1,2,3,4,5,5,6,7,8,9)

//求交集

scala>rdd3.intersection(rdd4)
    scala>rdd7.colleect	//查看结果
    rdd8:Array[Int] = Array(5)

练习五：

scala>val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2)))
    scala>val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
    //求jion
    scala>val rdd3 = rdd1.join(rdd2)	//返回一个新的RDD[(String,(Int,Int))]
    scala>rdd3.sv=aveAsText("hdfs://hadoop02/testjion")	//也可以存储
    scala>rdd3.collect	//查看结果
    Array[(String,Int)] = Array(("tom", 1), ("jerry", 3), ("kitty", 2),("jerry", 2), ("tom", 1), ("shuke", 2))
    //求并集
    scala>val rdd4 = rdd1 union rdd2
    //按key进行分组
    scala>rdd4.groupByKey
    scala>rdd4.collect	//结果
    key相同的会分到一组

练习六：

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5)) 
    //reduce聚合 
    val rdd2 = rdd1.reduce(_ + _) 
    rdd2.collect 
    //一次写完
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))。reduce(_ + _).collect

练习七：

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3),("kitty", 2),("shuke", 1))) 
    val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3),("shuke", 2), ("kitty", 5))) 
    val rdd3 = rdd1.union(rdd2) 
    //按key进行聚合 
    val rdd4 = rdd3.reduceByKey(_ + _) 
    rdd4.collect 
    //按value的降序排序 
    val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t => (t._2, t._1)) 
    rdd5.collect

五、mapPartitionsWithIndex 一次拿出一个分区（分区中并没有数据，而是记录要读取那些数据，真正生成的Task会读取多条数据），并且可以将分区的编号取出来

功能：取分区中对应的数据时，还可以将分区的编号取出来，这样就可以知道数据时属于哪一个分区的（哪个分区对应的Task的数据）

实战：

scala> val rdd = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)	//2代表两个分区
    需求：现在我先知道第一个分区和第二个分区对应哪些数据呢？调用mapPartitionsWithIndex就可以知道了
    //该函数功能时将对应的分区的 数据 取出来，并且带上分区编号
    scala> val rdd = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
    scala> val func = (index:Int, it:Iterator[Int]) => {
    it.map(e => $"part: $index, ele: $e")}	//（e代表每个元素）
    scala> rdd.map
    scala。 rdd.mapPartitions
    scala> val rdd2 = rdd.mapPartitionsWithIndex(func)
    scala> rdd2.collect		//查看rdd2有哪些内容
    这样得出来的一个分区有4个元素，另一个分区有5个元素

六、RDD相关概念

RDD 在计算的时候，每个分区都会起一个 task，所以 rdd 的分区数目决定了总的的 task 数目。

申请的计算节点（Executor）数目和每个计算节点核数，决定了你同一时刻可以并行执行的 task。

比如的 RDD 有 100 个分区，那么计算的时候就会生成 100 个 task，你的资源配置为 10 个计算节点，每个两 2 个核，同一时刻可以并行的 task 数目为20，计算这个 RDD 就需要 5 个轮次。

如果计算资源不变，你有 101 个 task 的话，就需要 6 个轮次，在最后一轮中，只有一个 task 在执行，其余核都在空转。

如果资源不变，你的 RDD 只有 2 个分区，那么同一时刻只有 2 个 task运行，其余 18 个核空转，造成资源浪费。这就是在 spark 调优中，增大 RDD分区数目，增大任务并行度的做法。

七、 键值对 RDD

键值对 RDD 是 Spark 中许多操作所需要的常见数据类型。

1）键值对RDD 的转化操作 ，

2）聚合操作 ：当数据集以键值对形式组织的时候，聚合具有相同键的元素进行一些统计是很常见的操作。

reduceByKey() 与 reduce() 相当类似;它们都接收一个函数，并使用该函数对值进行合并。 reduceByKey() 会为数据集中的每个键进行并行的归约操作，每个归约操作会将键相同的值合并起来。

foldByKey() 则与 fold() 相当类似，它们都使用一个与 RDD 和合并函数中的数据类型相同的零值作为初始值

combineByKey() 是最为常用的基于键进行聚合的函数。大多数基于键聚合的函数都是用它实现的。

八、数据分组

如果数据已经以预期的方式提取了键，groupByKey() 就会使用 RDD中的键来对数据进行分组。

九、链接

连接主要用于多个 Pair RDD 的操作，连接方式多种多样:右外连接、左外连接、交 叉连接以及内连接。

join 、leftOuterJoin()、rightOuterJoin()

十、数据排序

sortByKey() 函数接收一个叫作 ascending 的参数，表示我们是否想要让结果按升序排序(默认值为 true)。

十一、键值对RDD的数据分区

1）获取 RDD 的分区方式 ：可以通过使用 RDD 的 partitioner 属性来获取 RDD 的分区方式。 会返回一个scala.Option 对象， 通过 get 方法获取其中的值。

2）从分区中获益的操作 ：能够从数据分区中获得性能提升的操作有 cogroup()、 groupWith()、join()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()、 combineByKey() 以及 lookup()等。

十二、文本文件输入输出

当我们将一个文本文件读取为 RDD 时，输入的每一行 都会成为 RDD的一个元素。也可以将多个完整的文本文件一次性读取为一个 pair RDD， 其中键是文件名，值是文件内容。

val input = sc.textFile("./README.md")

通过 wholeTextFiles()对于大量的小文件读取效率比较高，大文件效果没有那么高。

Spark 通过 saveAsTextFile() 进行文本文件的输出，该方法接收一个路径，并将 RDD 中的内容都输入到路径对应的文件中。

十三、数据库的输入输出

关系型数据库连接 ，支持通过Java JDBC访问关系型数据库。需要通过JdbcRDD进行，示例如下：

Mysql读取：

def main (args: Array[String] ) { 
      val sparkConf = new SparkConf ().setMaster ("local[2]").setAppName 
    ("JdbcApp") 
      val sc = new SparkContext (sparkConf) 
     
      val rdd = new org.apache.spark.rdd.JdbcRDD ( 
        sc, 
        () => { 
          Class. forName ("com.mysql.jdbc.Driver").newInstance() 
          java.sql.DriverManager.getConnection ("jdbc:mysql://master01:3306/rdd", 
    "root", "hive") 
        }, 
        "select * from rddtable where id >= ? and id <= ?;", 
        1, 
        10, 
        1, 
        r => (r.getInt(1), r.getString(2))) 
     
      println (rdd.count () )
       rdd.foreach (println (_) ) 
      sc.stop () 
    }

mql写入：

def main(args: Array[String]) { 
      val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp") 
      val sc = new SparkContext(sparkConf) 
      val data = sc.parallelize( List ("Female", "Male","Female")) 
     
      data.foreachPartition(insertData) 
    } 
     
    def insertData(iterator: Iterator[String]): Unit = { 
    Class.forName ("com.mysql.jdbc.Driver").newInstance() 
      val conn = 
    java.sql.DriverManager.getConnection("jdbc:mysql://master01:3306/rdd", "root", 
    "hive") 
      iterator.foreach(data => { 
        val ps = conn.prepareStatement("insert into rddtable(name) values (?)") 
        ps.setString(1, data)  
        ps.executeUpdate() 
      }) 
    }

十四、HBase 数据库

由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat 类的实现，Spark 可以通过Hadoop 输入格式访问 HBase。这个输入格式会返回键值对数据，其中键的类型为 org.apache.hadoop.hbase.io.ImmutableBytesWritable，而值的类型为org.apache.hadoop.hbase.client.Result。

HBase 读取：

def main(args: Array[String]) { 
      val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp") 
      val sc = new SparkContext(sparkConf) 
     
      val conf = HBaseConfiguration.create() 
      //HBase 中的表名 
      conf.set(TableInputFormat.INPUT_TABLE, "fruit") 
     
      val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], 
        classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], 
        classOf[org.apache.hadoop.hbase.client.Result]) 
     
      val count = hBaseRDD.count() 
      println("hBaseRDD RDD Count:"+ count) 
      hBaseRDD.cache() 
      hBaseRDD.foreach { 
        case (_, result) => 
          val key = Bytes.toString(result.getRow) 
          val name = Bytes.toString(result.getValue("info".getBytes, 
          "name".getBytes)) 
          val color = Bytes.toString(result.getValue("info".getBytes, 
    "color".getBytes)) 
          println("Row key:" + key + " Name:" + name + " Color:" + color) 
      } 
      sc.stop() 
    }

HBase 写入：

def main(args: Array[String]) { 
      val sparkConf = new SparkConf().setMaster("local[2]").setAppName("HBaseApp") 
      val sc = new SparkContext(sparkConf) 
     
      val conf = HBaseConfiguration.create() 
      val jobConf = new JobConf(conf) 
      jobConf.setOutputFormat( classOf [TableOutputFormat]) 
      jobConf.set(TableOutputFormat.OUTPUT_TABLE, "fruit_spark") 
     
      val fruitTable = TableName.valueOf("fruit_spark") 
      val tableDescr = new HTableDescriptor(fruitTable) 
      tableDescr.addFamily(new HColumnDescriptor("info".getBytes)) 
     
      val admin = new HBaseAdmin(conf) 
      if (admin.tableExists(fruitTable)) { 
        admin.disableTable(fruitTable) 
        admin.deleteTable(fruitTable) 
      } 
      admin.createTable(tableDescr) 
     
      def convert(triple: (Int, String, Int)) = { 
        val put = new Put(Bytes.toBytes(triple._1)) 
        put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("name"), 
    Bytes.toBytes(triple._2)) 
        put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("price"), 
    Bytes.toBytes(triple._3)) 
        (new ImmutableBytesWritable, put) 
      } 
      val initialRDD = sc.parallelize(List((1,"apple",11), (2,"banana",12), 
    (3,"pear",13))) 
      val localData = initialRDD.map(convert) 
     
      localData.saveAsHadoopDataset(jobConf)