前言

Spark计算框架为了能够进行高并发和高吞吐的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

RDD : 弹性分布式数据集

累加器：分布式共享只写变量

广播变量：分布式共享只读变量

接下来我们一起看看这三大数据结构是如何在数据处理中使用的。

分布式计算模拟--客户端向服务器发送计算请求

package test

import java.io.{ObjectOutputStream, OutputStream}
import java.net.Socket

object Driver {
  def main(args: Array[String]): Unit = {
    //连接服务器
    val client = new Socket("localhost", 9999)
    val out: OutputStream = client.getOutputStream
    val objectOut = new ObjectOutputStream(out)
    val task = new Task()
    objectOut.writeObject(task)
    objectOut.flush()
    objectOut.close()
    client.close()
    println("客户端数据发送完毕")

  }

}

package test

import java.io.{InputStream, ObjectInputStream}
import java.net.{ServerSocket, Socket}

object Executor {
  def main(args: Array[String]): Unit = {
    //启动服务器，接收数据
    val server = new ServerSocket(9999)
    println("服务器启动，等待接收数据")
    //等待客户端的连接
    val client: Socket = server.accept()
    val in: InputStream = client.getInputStream
    //对象输入流
    val objin = new ObjectInputStream(in)
    val task: Task = objin.readObject().asInstanceOf[Task]
    val ints:List[Int] = task.compute()
    println("计算节点计算的结果为: "+ints)
    objin.close()
    client.close()
    server.close()
  }

}

package test

class Task extends Serializable {
  val datas=List(1,2,3,4)
//  val logic=(num:Int)=>{num*2}
  val logic:(Int)=>Int=_*2
  //计算
  def compute()={
    datas.map(logic)
  }


}

分布式计算模拟--数据结构和分布式计算

package test

import java.io.{ObjectOutputStream, OutputStream}
import java.net.Socket

object Driver {
  def main(args: Array[String]): Unit = {
    //连接服务器
    val client1 = new Socket("localhost", 9999)
    val client2 = new Socket("localhost", 8888)
    val task = new Task()
    val out1: OutputStream = client1.getOutputStream
    val objectOut1 = new ObjectOutputStream(out1)

    val subTask = new SubTask()
    subTask.logic=task.logic
    subTask.datas=task.datas.take(2)


    objectOut1.writeObject(subTask)
    objectOut1.flush()
    objectOut1.close()
    client1.close()
    println("客户端数据发送完毕")
    val out2: OutputStream = client2.getOutputStream
    val objectOut2 = new ObjectOutputStream(out2)

    val subTask1 = new SubTask()
    subTask1.logic = task.logic
    subTask1.datas = task.datas.takeRight(2)


    objectOut2.writeObject(subTask1)
    objectOut2.flush()
    objectOut2.close()
    client2.close()
    println("客户端数据发送完毕")

  }

}

package test

import java.io.{InputStream, ObjectInputStream}
import java.net.{ServerSocket, Socket}

object Executor {
  def main(args: Array[String]): Unit = {
    //启动服务器，接收数据
    val server = new ServerSocket(9999)
    println("服务器启动，等待接收数据")
    //等待客户端的连接
    val client: Socket = server.accept()
    val in: InputStream = client.getInputStream
    //对象输入流
    val objin = new ObjectInputStream(in)
    val task:SubTask = objin.readObject().asInstanceOf[SubTask]
    val ints:List[Int] = task.compute()
    println("计算节点[9999]计算的结果为: "+ints)
    objin.close()
    client.close()
    server.close()
  }

}

package test

import java.io.{InputStream, ObjectInputStream}
import java.net.{ServerSocket, Socket}

object Executor2 {
  def main(args: Array[String]): Unit = {
    //启动服务器，接收数据
    val server = new ServerSocket(8888)
    println("服务器启动，等待接收数据")
    //等待客户端的连接
    val client: Socket = server.accept()
    val in: InputStream = client.getInputStream
    //对象输入流
    val objin = new ObjectInputStream(in)
    val task: SubTask= objin.readObject().asInstanceOf[SubTask]
    val ints:List[Int] = task.compute()
    println("计算节点[8888]计算的结果为: "+ints)
    objin.close()
    client.close()
    server.close()
  }

}

package test

class SubTask extends Serializable {
  var datas:List[Int]=_
  var logic: (Int) => Int = _

  //计算
  def compute() = {
    datas.map(logic)
  }

}

package test

class Task extends Serializable {
  val datas=List(1,2,3,4)
//  val logic=(num:Int)=>{num*2}
  val logic:(Int)=>Int=_*2
  //计算
  def compute()={
    datas.map(logic)
  }


}

1、RDD概述

1.1 什么是RDD

RDD(Resilient Distributed Dataset)叫弹性分布式数据集，是Spark中对于分布式数据集的抽象。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

1.2 RDD五大特性

1、一组分区，即是数据集的基本组成单位，标记数据是哪个分区的

2、一个计算每个分区的函数

3、RDD之间的依赖关系

4、一个Partitioner，即RDD的分片函数：控制分区的数据流向（键值对）

5、一个列表，储存存取每个Partition的优先位置（prefered Location）。如果节点和分区个数不对应优先把分区设置在那个节点。移动数据不如移动计算，除非资源不够。

RDD--IO基本实现原理

说明：图二加了一个缓存区，等缓冲区满了在一起打印出来，类似批处理，提高了性能

RDD和IO之间的关系

RDD特点

1.弹性
存储的弹性：内存与磁盘的自动切换；
容错的弹性：数据丢失可以自动恢复；
计算的弹性：计算出错重试机制；
分片的弹性：可根据需要重新分片。
2.分布式：数据存储在大数据集群不同节点上
数据集：RDD封装了计算逻辑，并不保存数据
数据抽象：RDD是一个抽象类，需要子类具体实现
不可变：RDD封装了计算逻辑，是不可以改变的，想要改变，只能产生新的RDD，在新的RDD里面封装计算逻辑
可分区、并行计算

核心属性

分区列表

RDD数据结构中存在分区列表，用于执行任务时并行计算，是实现分布式计算的重要属性。

分区计算函数

Spark在计算时，是使用分区函数对每一个分区进行计算

RDD之间的依赖关系

RDD是计算模型的封装，当需求中需要将多个计算模型进行组合时，就需要将多个RDD建立依赖关系

分区器（可选）

当数据为KV类型数据时，可以通过设定分区器自定义数据的分区

首选位置（可选）

计算数据时，可以根据计算节点的状态选择不同的节点位置进行计算

计算向数据靠拢

执行原理

从计算的角度来讲，数据处理过程中需要计算资源（内存 & CPU）和计算模型（逻辑）。执行时，需要将计算资源和计算模型进行协调和整合。

Spark框架在执行时，先申请资源，然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。最后得到计算结果。

RDD是Spark框架中用于数据处理的核心模型，接下来我们看看，在Yarn环境中，RDD的工作原理:

1) 启动Yarn集群环境

2) Spark通过申请资源创建调度节点和计算节点

3) Spark框架根据需求将计算逻辑根据分区划分成不同的任务

4) 调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

从以上流程可以看出RDD在整个流程中主要用于将逻辑进行封装，并生成Task发送给Executor节点执行计算，接下来我们就一起看看Spark框架中RDD是具体是如何进行数据处理的。

2、RDD编程

2.1 RDD的创建

在Spark中创建RDD的创建方式可以分为三种：

1、从集合中创建

2、从外部储存创建

3、从其他RDD创建

1) 从集合（内存）中创建RDD

从集合中创建RDD，Spark主要提供了两个方法：parallelize和makeRDD

//1)从集合（内存）中创建RDD
//parallelize:并行

从底层代码实现来讲，makeRDD方法其实就是parallelize方法

package com.atguigu.rdd.builder

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

object RDD_Memory {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)
    //1)从集合（内存）中创建RDD
    //parallelize:并行
    val seq: Seq[Int] = Seq[Int](1, 2, 3, 4)
    val rdd: RDD[Int] = sc.parallelize(seq)
//    val rdd1 = sc.parallelize(
//      List(1, 2, 3, 4)
//    )
    val rdd1: RDD[Int] = sc.makeRDD(seq)
    rdd.collect().foreach(println)
    rdd1.collect().foreach(println)
    sc.stop()
  }

}

2) 从外部存储（文件）创建RDD

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

textFile:以行为单位来读取数据，读取的数据都是字符串
wholeTextFile:以文件为单位读取数据
读取的结果表示为元组，第一个元素表示文件路径，第二个元素表示文件内容

package com.atguigu.rdd.builder

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

object RDD_file {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sparkContext = new SparkContext(sparkConf)
    //path路径可以是文件的具体路径，也可以是目录名称
    //也可以是通配符*
    //还可以是分布式存储系统路径  "hdfs://node1:8020/test.txt"
    val fileRDD: RDD[String] = sparkContext.textFile("F:\\SparkCore代码\\Spark-core\\input")
    val rdd1: RDD[String] = sparkContext.textFile("datas/1*.txt")
    fileRDD.collect().foreach(println)
    sparkContext.stop()
  }

}

RDD并行度与分区

默认情况下，Spark可以将一个作业切分多个任务后，发送给Executor节点并行计算，而能够并行计算的任务数量我们称之为并行度。这个数量可以在构建RDD时指定。记住，这里的并行执行的任务数量，并不是指的切分任务的数量，不要混淆了。

package com.atguigu.rdd.builder

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object RDD_par {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    sparkConf.set("spark.default.parallelism","8")
    val sc = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    //将处理的数据保存成分区文件
    rdd.saveAsTextFile("output")
    sc.stop()
  }

}

分区数据的分配

读取内存数据时，数据可以按照并行度的设定进行数据的分区操作，数据分区规则的

var rdd=sc.makeRDD(List(1,2,3,4),3)

分区结果：
1
2
[3,4]

var rdd=sc.makeRDD(List(1,2,3,4,5),3)

分区结果：
1
[2,3]
[4,5]

读取内存数据时，数据可以按照并行度的设定进行数据的分区操作，数据分区规则的

def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
  (0 until numSlices).iterator.map { i =>
    val start = ((i * length) / numSlices).toInt
    val end = (((i + 1) * length) / numSlices).toInt
    (start, end)
  }
}

0=>0,1 1

1=>1,3 2

2=>3,5

读取文件数据时，数据是按照Hadoop文件读取的规则进行切片分区，而切片规则和数据读取的规则有些差异，具体Spark核心源码如下

textFile可以将文件作为数据处理的数据源，默认也可以设定分区

minPartitions：最小分区数量

match.min(defaultParallelism,2)

totalSize=7

goalSize(每个分区放三个字节)=7/2=3(byte)

//7/3=2...1(1.1)+1=3个分区

public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {

    long totalSize = 0;                           // compute total size
    for (FileStatus file: files) {                // check we have valid files
      if (file.isDirectory()) {
        throw new IOException("Not a file: "+ file.getPath());
      }
      totalSize += file.getLen();
    }

    long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
    long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
      FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);
      
    ...
    
    for (FileStatus file: files) {
    
        ...
    
    if (isSplitable(fs, path)) {
          long blockSize = file.getBlockSize();
          long splitSize = computeSplitSize(goalSize, minSize, blockSize);

          ...

  }
  protected long computeSplitSize(long goalSize, long minSize,
                                       long blockSize) {
    return Math.max(minSize, Math.min(goalSize, blockSize));
  }

RDD转换算子

RDD根据数据处理方式的不同将算子整体上分为Value类型、双Value类型和Key-Value类型

l Value类型

1) map

函数签名

def map[U: ClassTag](f: T => U): RDD[U]

函数说明

将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。

package oerator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object spark_map {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount3")
    val sc = new SparkContext(conf)
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
    val mapRDD: RDD[Int] = rdd.map(
      num => num * 2
    )
    mapRDD.collect().foreach(println)
  }

}

v 小功能：从服务器日志数据apache.log中获取用户请求URL资源路径

package oerator.transform

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

object spark_mapprait {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount3")
    val sc = new SparkContext(conf)
    val rdd: RDD[String] = sc.textFile("F:\\SparkCore代码\\Spark-core\\input\\apache.log")
    //长的字符串变成短的字符串
    val mapRDD: RDD[String] = rdd.map(
      line => {
        val datas: Array[String] = line.split(" ")
        datas(6)
      }
    )
    mapRDD.collect().foreach(println)
  }

}

并行计算效果

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object spark_map1 {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //map算子: 将数据集中的元素挨个经过指定的转换规则进行处理.

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 1)

    val mapRDD: RDD[Int] = listRDD.map(
      (num: Int) => {
        println(">>>>>>> " + num)
        num * 2
      }
    ).map(
      num => {
        println("******* " + num + 2)
        num + 2
      }

    )
    //val mapRDD: RDD[Int] = listRDD.map(_ * 2)

    mapRDD.collect() //.foreach(println)

    //mapRDD.saveAsTextFile("output")


    //集合的方法操作是单点操作
    //    val list = List(1,2,3,4)
    //    val result: List[Int] = list.map(_ * 2)
    //    println(result)

    sc.stop()

  }

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object spark_map1 {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //map算子: 将数据集中的元素挨个经过指定的转换规则进行处理.

    val listRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    val mapRDD: RDD[Int] = listRDD.map(
      (num: Int) => {
        println(">>>>>>> " + num)
        num * 2
      }
    ).map(
      num => {
        println("******* " + num + 2)
        num + 2
      }

    )
    //val mapRDD: RDD[Int] = listRDD.map(_ * 2)

    mapRDD.collect() //.foreach(println)

    //mapRDD.saveAsTextFile("output")


    //集合的方法操作是单点操作
    //    val list = List(1,2,3,4)
    //    val result: List[Int] = list.map(_ * 2)
    //    println(result)

    sc.stop()

  }

}

2) mapPartitions

Ø 函数签名

def mapPartitions[U: ClassTag](

f: Iterator[T] => Iterator[U],

preservesPartitioning: Boolean = false): RDD[U]

函数说明

将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据。

val dataRDD1: RDD[Int] = dataRDD.mapPartitions(

datas => {

datas.filter(_==2)

}

)

v 小功能：获取每个数据分区的最大值

思考一个问题：map和mapPartitions的区别？

数据处理角度

Map算子是分区内一个数据一个数据的执行，类似于串行操作。而mapPartitions算子是以分区为单位进行批处理操作。

功能的角度

Map算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。MapPartitions算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，所以可以增加或减少数据

性能的角度

Map算子因为类似于串行操作，所以性能比较低，而是mapPartitions算子类似于批处理，所以性能较高。但是mapPartitions算子会长时间占用内存，那么这样会导致内存可能不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用map操作。

完成比完美更重要

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object spark_mapPartitions {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)


    rdd.mapPartitions(
      //mapPartitions : 将每个分区的数据整体进行处理.
      list => {
        // println("************* " + list.mkString(" , ") )
//        List(list.max).iterator
        println(">>>>>>>>>>>>>>>>>>>>")
        list.map(_*2)
      }
    ).collect().foreach(println)


  }

}

v 小功能：获取每个数据分区的最大值

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object spark_mapPartitions1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)
    //需求: 求每个分区的最大值
    //mapPartitions : 将每个分区的数据整体进行处理.
    //[1,2],[3,4]
    //[2].[4]
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    val mpRDD: RDD[Int] = rdd.mapPartitions(
      iter => {
        List(iter.max).iterator
      }
    )
    mpRDD.foreach(println)
    sc.stop()

  }

}

3) mapPartitionsWithIndex

函数签名

def mapPartitionsWithIndex[U: ClassTag](

f: (Int, Iterator[T]) => Iterator[U],

preservesPartitioning: Boolean = false): RDD[U]

函数说明

将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。

val dataRDD1 = dataRDD.mapPartitionsWithIndex(

(index, datas) => {

datas.map(index, _)

}

)

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Spark_mapPartitionsWithIndex {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //需求: 获取0号分区的数据.
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)


    rdd.mapPartitionsWithIndex(
      (index,datas) => {
        if(index == 0 ) {  //第一分区
          datas
        }else{
          Nil.iterator //空迭代器
        }
      }
    ).collect().foreach(println)  //1 2

    sc.stop()


  }

}

object Spark_mapPartitionsWithIndex1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //需求: 获取每个元素在哪一个分区上
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
    rdd.mapPartitionsWithIndex(
      (index,iter)=>{
        //1, 2   3,4
        //(0,1)(2,2),(4,3)(6,3)
        iter.map(num=>{
          (index,num)
        })
      }
    ).foreach(println)
  }

}

v 小功能：获取1号分区的数据

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Spark_mapPartitionsWithIndex1 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)
    //需求: 获取1号分区的数据.
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    rdd.mapPartitionsWithIndex {
      case (1, datas) => datas
      case _ => Nil.iterator
    }.collect().foreach(println)  //3,4

    sc.stop()
  }

}

4) flatMap

函数签名

def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

函数说明

将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射

val dataRDD = sparkContext.makeRDD(List(

List(1,2),List(3,4)

),1)

val dataRDD1 = dataRDD.flatMap(

list => list

)

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Spark_flatmap {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //flatMap算子:
    //List(List(1,2),3,List(4,5))
    val listRDD = sc.makeRDD(List(List(1, 2), List(4, 5)))
    val flatRDD: RDD[Int] = listRDD.flatMap(
      list => {
        list
      }
    )
    flatRDD.foreach(println)  //1 2 3 4
    sc.stop()
  }

}

package oerator.transform

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

object Spark_flatmap1 {
  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)
    val stringRDD: RDD[String] = sc.makeRDD(List("Hello Spark", "Hello Scala", "Hello Spark", "Hello World"), 2)
     /*
     * s=>{
     * s.split(" ")
     * }
     * */
    stringRDD.flatMap(_.split(" ")).collect().foreach(println)

    sc.stop()

  }

}

v 小功能：将List(List(1,2),3,List(4,5))进行扁平化操作

package oerator.transform
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Spark_flatmapprait {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    //flatMap算子:
    //List(List(1,2),3,List(4,5))
    val listRDD = sc.makeRDD(List(List(1, 2), 3, List(4, 5)), 2)
    listRDD.flatMap(
      data=>{
        data match {
          case list:List[_]=>list  //如果是集合类型就返回集合
          case dat=>List(dat)   //不是集合变成集合
        }
      }
    ).collect().foreach(println)
//
//    listRDD.flatMap {
//      case ele: List[_] => ele
//      case ele: Int => List(ele)
//    }.collect().foreach(println)
    sc.stop()
  }

}

5) glom

函数签名

def glom(): RDD[Array[T]]

函数说明

将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变

val dataRDD : RDD[Int] = sparkContext.makeRDD(List(

1,2,3,4

),1)

val dataRDD1:RDD[Array[Int]] = dataRDD.glom()

v 小功能：计算所有分区最大值求和（分区内取最大值，分区间最大值求和）

6) groupBy

函数签名

def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

函数说明

将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为shuffle。极限情况下，数据可能被分在同一个分区中

一个组的数据在一个分区中，但是并不是说一个分区中只有一个组

val dataRDD = sparkContext.makeRDD(List(1,2,3,4),1)

val dataRDD1 = dataRDD.groupBy(

_%2

)

v 小功能：将List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。

v 小功能：从服务器日志数据apache.log中获取每个时间段访问量。

v 小功能：WordCount。

7) filter

函数签名

def filter(f: T => Boolean): RDD[T]

函数说明

将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。

当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

val dataRDD = sparkContext.makeRDD(List(

1,2,3,4

),1)

val dataRDD1 = dataRDD.filter(_%2 == 0)

v 小功能：从服务器日志数据apache.log中获取2015年5月17日的请求路径

Spark学习---day02、Spark核心编程（RDD概述、RDD编程（创建、分区规则、转换算子、Action算子））（一）

前言

分布式计算模拟--客户端向服务器发送计算请求

分布式计算模拟--数据结构和分布式计算

1、RDD概述

1.1 什么是RDD

1.2 RDD五大特性

RDD--IO基本实现原理

RDD和IO之间的关系

核心属性

执行原理

2、RDD编程

2.1 RDD的创建

RDD并行度与分区

分区数据的分配

RDD转换算子

1) map

2) mapPartitions

3) mapPartitionsWithIndex

4) flatMap

5) glom

6) groupBy

7) filter

热门文章

最新文章

相关课程

相关电子书

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Spark学习---day02、Spark核心编程（RDD概述、RDD编程（创建、分区规则、转换算子、Action算子））（一）

前言

分布式计算模拟--客户端向服务器发送计算请求

分布式计算模拟--数据结构和分布式计算

1、RDD概述

1.1 什么是RDD

1.2 RDD五大特性

RDD--IO基本实现原理

RDD和IO之间的关系

核心属性

执行原理

2、RDD编程

2.1 RDD的创建

RDD并行度与分区

分区数据的分配

RDD转换算子

1) map

2) mapPartitions

3) mapPartitionsWithIndex

4) flatMap

5) glom

6) groupBy

7) filter

热门文章

最新文章

相关课程

相关电子书