前言
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 小功能：计算所有分区最大值求和（分区内取最大值，分区间最大值求和）
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_glom12 { def main(args: Array[String]): Unit = { //glom:将同一分区的数据直接转化为相同类型的内存数组进行处理，分区不变 val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc = new SparkContext(sparkConf) val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2) //需求: 计算所有分区最大值求和（分区内取最大值，分区间最大值求和） val rdd1: RDD[Array[Int]] = rdd.glom() val maxRDD: RDD[Int] = rdd1.map(w => w.max) val result: Double = maxRDD.sum() maxRDD.collect().foreach(println) println(result) sc.stop() } }
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
)
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_group5 { 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, 6, 8, 10, 12, 14, 16), 2) // rdd.groupBy( // num=>{ // if(num%2==0){ // "偶数" // }else{ // "奇数" // } // } // ).collect().foreach(println) rdd.groupBy(_%2==0,2).collect().foreach(println) } }
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_groupBy { def main(args: Array[String]): Unit = { //groupBy:会将数据源中每个数据进行分组判断，根据返回的分组key进行分组 //相同的key值的数据会放置在一个组里 val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc = new SparkContext(sparkConf) val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2) def groupFuncation(num:Int)={ num%2 } val groupRDD: RDD[(Int, Iterable[Int])] = rdd.groupBy(groupFuncation) groupRDD.collect().foreach(println) } }
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_groupBy1 { def main(args: Array[String]): Unit = { //groupBy:会将数据源中每个数据进行分组判断，根据返回的分组key进行分组 //相同的key值的数据会放置在一个组里 val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc = new SparkContext(sparkConf) val rdd = sc.makeRDD(List("Hello", "Spark", "Scala", "Hadoop"),2) //把首写字母大写的放在一起 val groupRDD = rdd.groupBy(_.charAt(0)) groupRDD.collect().foreach(println) } }
v 小功能：将List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_groupBy2 { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc = new SparkContext(sparkConf) val rdd: RDD[String] = sc.makeRDD(List("Hello", "hive", "hbase", "Hadoop"), 2) val grouprdd: RDD[(Char, Iterable[String])] = rdd.groupBy(_.charAt(0)) grouprdd.collect().foreach(println) } }
v 小功能：从服务器日志数据apache.log中获取每个时间段访问量。
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} import java.text.SimpleDateFormat import java.util.Date object spark_groupBy3 { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc = new SparkContext(sparkConf) // 从服务器日志数据apache.log中获取每个时间段访问量 val rdd: RDD[String] = sc.textFile("F:\\SparkCore代码\\Spark-core\\input\\apache.log") val timeRDD: RDD[(String, Iterable[(String, Int)])] = rdd.map( line => { val datas: Array[String] = line.split(" ") val time: String = datas(3) val sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss") val data: Date = sdf.parse(time) val sdf1 = new SimpleDateFormat("HH") val hour: String = sdf1.format(data) (hour, 1) } ).groupBy(_._1) timeRDD.map{ case(hour,iter)=>{ (hour,iter.size) } }.collect().foreach(println) } }
方法二:
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_group4 { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val fileRDD: RDD[String] = sc.textFile("F:\\SparkCore代码\\Spark-core\\input\\apache.log") // line => (hour,1) fileRDD.map( line => { val logs: Array[String] = line.split(" ") val times: String = logs(3) val timesArr: Array[String] = times.split(":") (timesArr(1), 1) } // (hour,1) => ( hour, [(hour,1), (hour,1)...]) ).groupBy(_._1) .map(kv=>{ (kv._1,kv._2.size) }).collect().foreach(println) /* * * timeRDD.map{ * case(hour,iter)=>{ * (hour,iter.size) * } * }.collect.foreach(println)*/ // .mapValues(_.map(_._2).sum) // .collect().foreach(println) } }
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)
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_filter { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //filter算子: val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 6, 8, 10, 12, 14, 16), 2) val filterRDD: RDD[Int] = rdd.filter(num => num % 2 != 0) filterRDD.collect().foreach(println) } }
v 小功能：从服务器日志数据apache.log中获取2015年5月17日的请求路径
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_filter1 { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val fileRDD: RDD[String] = sc.textFile("F:\\SparkCore代码\\Spark-core\\input\\apache.log") fileRDD.filter( line => { val logs: Array[String] = line.split(" ") logs(3).startsWith("17/05/2015") } ).map( line => { val logs: Array[String] = line.split(" ") logs(logs.length - 1) } ).collect().foreach(println) sc.stop() } }
8)sample
def sample(
withReplacement: Boolean,
fraction: Double,
seed: Long = Utils.random.nextLong): RDD[T]
函数说明
根据指定的规则从数据集中抽取数据
val dataRDD = sparkContext.makeRDD(List(
1,2,3,4
),1)
// 抽取数据不放回（伯努利算法） // 伯努利算法：又叫0、1分布。例如扔硬币，要么正面，要么反面。 // 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要 // 第一个参数：抽取的数据是否放回，false：不放回 // 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取； // 第三个参数：随机数种子 val dataRDD1 = dataRDD.sample(false, 0.5) // 抽取数据放回（泊松算法） // 第一个参数：抽取的数据是否放回，true：放回；false：不放回 // 第二个参数：重复数据的几率，范围大于等于0.表示每一个元素被期望抽取到的次数 // 第三个参数：随机数种子 val dataRDD2 = dataRDD.sample(true, 2)
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_sample { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //sample算子: val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), 2) /* sample可以传递三个传参数: 1、 表示抽取的数据是否放回. true ： 放回， false: 不放回. 2、 如果是抽取不放回的情况，第二个参数表示抽取的几率，[0-1], 0表示都不抽取， 1 表示都抽取 如果是抽取放回的情况,第二个参数表示每个元素期望被抽取到的次数. 3、 种子 */ //rdd.sample(false,0.5).collect().foreach(println) //rdd.sample(false,0.5,1L).collect().foreach(println) //rdd.sample(true,2).collect().foreach(println) rdd.sample(true, 2, 1L).collect().foreach(println) sc.stop() } }
9) distinct
将数据集中重复的数据去重
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_distinct { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //distinct 算子: val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 1, 2, 4, 7, 8), 2) rdd.distinct().collect().foreach(println) sc.stop() } }
10）coalesce
根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
当spark程序中，存在过多的小任务的时候，可以通过coalesce方法，收缩合并分区，减少分区的个数，减小任务调度成本
package oerator.transform import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.rdd.RDD object spark_coalesce { 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, 5, 6), 3) //缩减分区 // coalesce缩减分区默认不走shuffle // 如果想要在缩减分区时同时进行shuffle操作，可以通过指定第二个参数为true来实现. val newRDD: RDD[Int] = rdd.coalesce(2) val newRDD: RDD[Int] = rdd.coalesce(2,true) newRDD.saveAsTextFile("output1") sc.stop() } }
11） repartition
该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition操作都可以完成，因为无论如何都会经shuffle过程。
package oerator.transform import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.rdd.RDD object spark_coalesce { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //distinct 算子: val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3) //缩减分区 // coalesce缩减分区默认不走shuffle // 如果想要在缩减分区时同时进行shuffle操作，可以通过指定第二个参数为true来实现. //val newRDD: RDD[Int] = rdd.coalesce(2) //val newRDD: RDD[Int] = rdd.coalesce(2,true) //newRDD.saveAsTextFile("output1") //扩大分区: // val newRDD: RDD[Int] = rdd.coalesce(6) // val newRDD: RDD[Int] = rdd.coalesce(6,true) // newRDD.saveAsTextFile("output3") // val newRDD: RDD[Int] = rdd.repartition(6) sc.stop() } }
12）sortby
该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为升序排列。排序后新产生的RDD的分区数与原RDD的分区数一致。中间存在shuffle的过程
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_sortBy { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //sortBy 算子: //val rdd: RDD[Int] = sc.makeRDD( List(8,6,2,7,4,3,1) ,2) //升序 、 降序 //rdd.sortBy( num => num , false ).collect().foreach(println) val rdd: RDD[(String, Int)] = sc.makeRDD(List(("1", 1), ("11", 2), ("2", 3)), 2) //第二个参数可以降序 val newRDD: RDD[(String, Int)] = rdd.sortBy(t => t._1.toInt,false) //字符串变数值 newRDD.collect().foreach(println) sc.stop() } }
13)交并差
package oerator.transform import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.rdd.RDD object spark_doublevalue { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //交集，并集和差集要求两个数据源数据类型保持一致 //拉链操作两个数据源的类型不一致 //两个数据源要求分区数量保持一致 val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3) val rdd2: RDD[Int] = sc.makeRDD(List(3, 4, 5, 6), 2) val rdd3: RDD[String] = sc.makeRDD(List("a", "b", "c", "d"), 2) //交集 rdd1.intersection(rdd2).collect().foreach(println) println("-----------") //并集 rdd1.union(rdd2).collect().foreach(println) println("-----------") //差集 rdd1.subtract(rdd2).collect().foreach(println) println("-----------") rdd2.subtract(rdd1).collect().foreach(println) //rdd1.intersection(rdd3) println("-----------") // Can only zip RDDs with same number of elements in each partition //Can't zip RDDs with unequal numbers of partitions //rdd1.zip(rdd2).collect().foreach(println) rdd2.zip(rdd3).collect().foreach(println) sc.stop() } }
14) partitionBy
将数据按照指定Partitioner重新进行分区。Spark默认的分区器是HashPartitioner
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{HashPartitioner, SparkConf, SparkContext} object spark_partitionBy { 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=>PairRDDFuncations val mapRDD: RDD[(Int, Int)] = rdd.map((_, 1)) mapRDD.partitionBy(new HashPartitioner(2)).saveAsTextFile("output") // val rdd = sc.makeRDD( // List( // ("a", 1), ("b", 1), ("c", 1), ("d", 1) // ), // 2) // //coalesce: 侧重于分区的数量 // //repartition:侧重于分区的数量 // //Partitioner:侧重于分区的数据 // rdd.partitionBy(new HashPartitioner(2)).saveAsTextFile("output") // sc.stop() } }
15) reduceByKey
可以将数据按照相同的Key对Value进行聚合
package oerator.transform import org.apache.spark.{SparkConf, SparkContext} object spark_reduceByKey { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val rdd = sc.makeRDD( List( ("a", 1), ("b", 1), ("a", 1), ("d", 1) ), 2) //wordcount rdd.reduceByKey(_ + _).collect().foreach(println) sc.stop() } }
16) groupByKey
将数据源的数据根据key对value进行分组
package oerator.transform import org.apache.spark.{SparkConf, SparkContext} object spark_groupByKey { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val rdd = sc.makeRDD( List( ("a", 1), ("b", 1), ("a", 1), ("d", 1) ), 2) rdd.groupByKey().collect().foreach(println) sc.stop() } }
reduceByKey和groupByKey的区别？
从shuffle的角度：reduceByKey和groupByKey都存在shuffle的操作，但是reduceByKey可以在shuffle前对分区内相同key的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高。 从功能的角度：reduceByKey其实包含分组和聚合的功能。groupByKey只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey
17) aggregateByKey
将数据根据不同的规则进行分区内计算和分区间计算
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_aggregateByKey { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) //需求: 分区内取相同key的最大值, 分区间求和 // 分区0 ： (a,1) (a,2) (b,3) ->分区内-> (a,2) (b,3) // 分区间求和： (a,8) (b,8) // 分区1 ： (b,4) (b,5) (a,6) ->分区内-> (b,5) (a,6) val rdd = sc.makeRDD( List( ("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6) ), 2) /* aggregateByKey算子有两个参数列表(柯里化) 1. 第一个参数列表 每个分区中每个不同的key第一次进行比较的初始值 2. 第二个参数列表,有两个参数 第一个参数 ： 分区内的计算逻辑 第二个参数 : 分区间的计算逻辑 */ rdd.aggregateByKey(0)( (x, y) => math.max(x, y), (x, y) => x + y ).collect().foreach(println) println("-----------------------------------") //需求: 分区内求和， 分区间求和 // reduceByKey => combineByKeyWithClassTag rdd.reduceByKey(_ + _).collect().foreach(println) println("-----------------------------------") //需求: 分区内求和, 分区间求和， 在分区内计算时给定初始值. //aggregateByKey => combineByKeyWithClassTag rdd.aggregateByKey(0)(_ + _, _ + _).collect().foreach(println) println("-----------------------------------") //如果分区内和分区间的计算逻辑是一样的， aggregateByKey可以简写成foldByKey // foldByKey => combineByKeyWithClassTag rdd.foldByKey(0)(_ + _).collect().foreach(println) sc.stop() } }
18) foldByKey
Ø 函数签名
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
函数说明
当分区内计算规则和分区间计算规则相同时，aggregateByKey就可以简化为foldByKey
19) join
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_join { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val rdd1 = sc.makeRDD( List( ("a", 1), ("b", 2), ("c", 3), ("a", 4), ("c", 9) ), 2 ) val rdd2 = sc.makeRDD( List( ("a", 5), ("d", 6), ("c", 7), ("c", 8) ), 2 ) // join ： 内连接 外连接(左 、 右 、 全) rdd1.join(rdd2).collect().foreach(println) println("----------------------------------") rdd1.leftOuterJoin(rdd2).collect().foreach(println) println("----------------------------------") rdd2.rightOuterJoin(rdd1).collect().foreach(println) println("----------------------------------") rdd1.fullOuterJoin(rdd2).collect().foreach(println) println("----------------------------------") // group + join rdd1.cogroup(rdd2).collect().foreach(println) sc.stop() } }
20) cogroup
在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD
package oerator.transform import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object spark_cogroup { def main(args: Array[String]): Unit = { val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val rdd1: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("b", 2))) val rdd2: RDD[(String, Int)] = sc.makeRDD(List(("a", 4), ("b", 5), ("c", 6))) //cogroup val cgRDD: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd1.cogroup(rdd2) cgRDD.collect().foreach(println) } }
案例实操
1) 数据准备
agent.log：时间戳，省份，城市，用户，广告，中间字段使用空格分隔。
2) 需求描述
统计出每一个省份每个广告被点击数量排行的Top3
3) 需求分析
4) 功能实现
package com.atguigu.rdd.builder import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object provAndadvClick { def main(args: Array[String]): Unit = { //统计出每一个省份每个广告被点击数量排行的Top3 val sparkConf: SparkConf = new SparkConf().setAppName("rdd").setMaster("local[*]") val sc: SparkContext = new SparkContext(sparkConf) val fileRDD: RDD[String] = sc.textFile("F:\\SparkCore代码\\Spark-core\\input\\agent.log") //1. ( (省份，广告), count) val mapRDD= fileRDD.map( line => { val lines: Array[String] = line.split(" ") ((lines(1), lines(lines.length - 1)), 1) } ) //2.将转换结构的数据进行分组聚合 //((省份,广告),1)=>((省份，广告),sum) val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(_ + _) //3.将聚合的结果进行结构转换 //((省份,广告),sum)=>(省份,(广告,sum)) val newMapRDD: RDD[(String, (String, Int))] = reduceRDD.map { case ((prv, ad), sum) => { (prv, (ad, sum)) } /* * reduceRDD.map{ * kv=>{ * kv._1 * } * } * */ } //4.分组 //[省份,[(广告A,sumA),(广告B,sumB)]) val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey() //5.将分组后的数据组内排序(降序)，取前三名 val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues( iter => { iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(3) } ) resultRDD.collect().foreach(println) } }
