transformation和action介绍

2023-02-25 15:34:09 浏览数 (1)

Spark支持两种RDD操作:transformation和action。transformation操作会针对已有的RDD创建一个新的RDD;而action则主要是对RDD进行最后的操作,比如遍历、reduce、保存到文件等,并可以返回结果给Driver程序。

例如,map就是一种transformation操作,它用于将已有RDD的每个元素传入一个自定义的函数,并获取一个新的元素,然后将所有的新元素组成一个新的RDD。而reduce就是一种action操作,它用于对RDD中的所有元素进行聚合操作,并获取一个最终的结果,然后返回给Driver程序。

transformation的特点就是lazy特性。lazy特性指的是,如果一个spark应用中只定义了transformation操作,那么即使你执行该应用,这些操作也不会执行。也就是说,transformation是不会触发spark程序的执行的,它们只是记录了对RDD所做的操作,但是不会自发的执行。只有当transformation之后,接着执行了一个action操作,那么所有的transformation才会执行。Spark通过这种lazy特性,来进行底层的spark应用执行的优化,避免产生过多中间结果。

action操作执行,会触发一个spark job的运行,从而触发这个action之前所有的transformation的执行。这是action的特性。

画图理解:

1.png

案例:统计文件每行出现的次数

这里通过一个之前学习过的案例,统计文件字数,来讲解transformation和action。

// 这里通过textFile()方法,针对外部文件创建了一个RDD,lines,但是实际上,程序执行到这里为止,spark.txt文件的数据是不会加载到内存中的。lines,只是代表了一个指向spark.txt文件的引用。

val lines = sc.textFile("spark.txt")

// 这里对lines RDD进行了map算子,获取了一个转换后的lineLengths RDD。但是这里连数据都没有,当然也不会做任何操作。lineLengths RDD也只是一个概念上的东西而已。

val lineLengths = lines.map(line => line.length)

// 之列,执行了一个action操作,reduce。此时就会触发之前所有transformation操作的执行,Spark会将操作拆分成多个task到多个机器上并行执行,每个task会在本地执行map操作,并且进行本地的reduce聚合。最后会进行一个全局的reduce聚合,然后将结果返回给Driver程序。

val totalLength = lineLengths.reduce(_ _)

案例:统计文件每行出现的次数

Spark有些特殊的算子,也就是特殊的transformation操作。比如groupByKey、sortByKey、reduceByKey等,其实只是针对特殊的RDD的。即包含key-value对的RDD。而这种RDD中的元素,实际上是scala中的一种类型,即Tuple2,也就是包含两个值的Tuple。

在scala中,需要手动导入Spark的相关隐式转换,import org.apache.spark.SparkContext._。然后,对应包含Tuple2的RDD,会自动隐式转换为PairRDDFunction,并提供reduceByKey等方法。

val lines = sc.textFile("hello.txt")

val linePairs = lines.map(line => (line, 1))

val lineCounts = linePairs.reduceByKey(_ _)

lineCounts.foreach(lineCount => println(lineCount._1 " appears " llineCount._2 " times."))

Java版本的案例:

代码语言:javascript复制
/**

* 统计每行出现的次数

* @author Administrator

*

*/

public class LineCount {

​public static void main(String[] args) {

​​//创建sparkConf 
​​SparkConf conf = newSparkConf().setAppName("LineCount").setMaster("local");

​​//创建JAVASparkContext
​​JavaSparkContext sc = new JavaSparkContext(conf);

​​//创建初始化RDD lines
​​JavaRDD<String> lines = sc.textFile("C:\Users\zhang\Desktop\hello.txt");

// 对lines RDD执行mapToPair算子,将每一行映射为(line, 1)的这种key-value对的格式
// 然后后面才能统计每一行出现的次数
JavaPairRDD<String, Integer> pairs = lines.mapToPair(newPairFunction<String, String, Integer>() {

   ​​​private static final long serialVersionUID = 1L;

   @Override
   ​​​public Tuple2<String, Integer> call(String t) throwsException {

    ​​​​return new Tuple2<String, Integer>(t, 1);

​​​}

​​});

​​// 对pairs RDD执行reduceByKey算子,统计出每一行出现的总次数
​JavaPairRDD<String, Integer> lineCounts =pairs.reduceByKey(

​​​new Function2<Integer, Integer, Integer>() {

​​​@Override
public Integer call(Integer v1, Integer v2) throwsException {
// TODO Auto-generated method stub

 return v1   v2;
​​​}
​​});

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

​​​private static final long serialVersionUID = 1L;

@Override
​​​public void call(Tuple2<String, Integer> t) throwsException {

  System.out.println(t._1   " appers "   t._2   " times.");
}
});

sc.close();
}
}

Scala版本

代码语言:javascript复制
object LineFile {

def main(args: Array[String]) { 

val conf = newSparkConf().setAppName("LineFile").setMaster("local")

val sc = new  SparkContext(conf)

val lines = sc.textFile("C:\Users\zhang\Desktop\hello.txt", 1)

val pairs = lines.map { line => (line,1) }

val lineCounts = pairs.reduceByKey(_   _)

lineCounts.foreach(lineCount => println(lineCount._1   " appears "  lineCount._2   " times."))
}
}

0 人点赞