6、transformation和action1

我就是我 2021-11-26 18:36 247阅读 0赞

**一、transformation和action入门**

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的特性。

[![image][]][image 1]

2、案例：统计文件字数

这里通过一个之前学习过的案例，统计文件字数，来讲解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(_ + _)

4、案例：统计文件每行出现的次数

Spark有些特殊的算子，也就是特殊的transformation操作。比如groupByKey、sortByKey、reduceByKey等，其实只是针对特殊的RDD的。即包含key-value对的RDD。
 而这种RDD中的元素，实际上是scala中的一种类型，即Tuple2，也就是包含两个值的Tuple。
 
 在scala中，需要手动导入Spark的相关隐式转换，import org.apache.spark.SparkContext._。然后，对应包含Tuple2的RDD，会自动隐式转换为PairRDDFunction，并提供
 reduceByKey等方法。
 
 
 
 -----------------java实现--------------------
 
 package cn.spark.study.core;
 
 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.Function2;
 import org.apache.spark.api.java.function.PairFunction;
 import org.apache.spark.api.java.function.VoidFunction;
 
 import scala.Tuple2;
 
 /**
 * 统计每行出现的次数
 * 
 * @author bcqf
 *
 */
 
 public class LineCount {
 public static void main(String[] args) {
 // 创建SparkConf
 SparkConf sparkConf = new SparkConf().setAppName("LineCountJava").setMaster("local");
 
 // 创建JavaSparkContext
 JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
 
 // 创建初始RDD，linesRDD 每个元素是一行文本
 JavaRDD<String> linesRDD = javaSparkContext.textFile("D:\\test-file\\hello.txt");
 
 // 对linesRDD执行mapToPair算子，将每一行映射为（lines，1）这种key-value对的格式，然后才能统计每一行出现的次数
 JavaPairRDD<String, Integer> pairsRDD = linesRDD.mapToPair(new PairFunction<String, String, Integer>() {
 @Override
 public Tuple2<String, Integer> call(String s) throws Exception {
 return new Tuple2<String, Integer>(s, 1);
 }
 
 });
 
 // 对pairsRDD执行reduceByKey算子，统计出每一行出现的总次数
 JavaPairRDD<String, Integer> linesCount = pairsRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
 @Override
 public Integer call(Integer integer, Integer integer2) throws Exception {
 return integer + integer2;
 }
 
 });
 
 // 执行一个action操作，foreach，打印出每一行出现的总次数
 linesCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
 @Override
 public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
 System.out.println(stringIntegerTuple2._1 + " appears " + stringIntegerTuple2._2 + " times");
 }
 
 });
 
 // 关闭javaSparkContext
 javaSparkContext.close();
 }
 }
 
 
 
 
 
 ----------------scala实现--------------------
 
 package cn.spark.study.core
 
 import org.apache.spark.SparkConf
 import org.apache.spark.SparkContext
 
 object LineCount {
 def main(args: Array[String]) {
 val conf = new SparkConf().setAppName("LineCount").setMaster("local")
 
 val sc = new SparkContext(conf)
 
 val lines = sc.textFile("D:\\test-file\\hello.txt", 1)
 
 val pairs = lines.map { line => (line, 1)}
 
 val lineCounts = pairs.reduceByKey {_ + _}
 
 lineCounts.foreach(lineCount => println(lineCount._1 + " appears " + lineCount._2 + " times."))
 }
 }

5、常用transformation

<table> 
 <tbody> 
 <tr> 
 <td>操作</td> 
 <td>介绍</td> 
 </tr> 
 <tr> 
 <td>map </td> 
 <td>将RDD中的每个元素传入自定义函数，获取一个新的元素，然后用新的元素组成新的RDD </td> 
 </tr> 
 <tr> 
 <td>filter </td> 
 <td>对RDD中每个元素进行判断，如果返回true则保留，返回false则剔除。 </td> 
 </tr> 
 <tr> 
 <td>flatMap </td> 
 <td>与map类似，但是对每个元素都可以返回一个或多个新元素。 </td> 
 </tr> 
 <tr> 
 <td>gropuByKey</td> 
 <td>根据key进行分组，每个key对应一个Iterable&lt;value&gt; </td> 
 </tr> 
 <tr> 
 <td>reduceByKey </td> 
 <td>对每个key对应的value进行reduce操作。</td> 
 </tr> 
 <tr> 
 <td>sortByKey </td> 
 <td>对每个key对应的value进行排序操作。 </td> 
 </tr> 
 <tr> 
 <td>join </td> 
 <td>对两个包含&lt;key,value&gt;对的RDD进行join操作，每个key join上的pair，都会传入自定义函数进行处理。 </td> 
 </tr> 
 <tr> 
 <td>cogroup </td> 
 <td>同join，但是是每个key对应的Iterable&lt;value&gt;都会传入自定义函数进行处理。 </td> 
 </tr> 
 </tbody> 
</table>

6、常用action

<table> 
 <tbody> 
 <tr> 
 <td>操作</td> 
 <td>介绍 </td> 
 </tr> 
 <tr> 
 <td>reduce </td> 
 <td>将RDD中的所有元素进行聚合操作。第一个和第二个元素聚合，值与第三个元素聚合，值与第四个元素聚合，以此类推。 </td> 
 </tr> 
 <tr> 
 <td>collect </td> 
 <td>将RDD中所有元素获取到本地客户端; </td> 
 </tr> 
 <tr> 
 <td>count </td> 
 <td>获取RDD元素总数。 </td> 
 </tr> 
 <tr> 
 <td>take(n) </td> 
 <td>获取RDD中前n个元素。 </td> 
 </tr> 
 <tr> 
 <td>saveAsTextFile </td> 
 <td>将RDD元素保存到文件中，对每个元素调用toString方法 </td> 
 </tr> 
 <tr> 
 <td>countByKey</td> 
 <td>对每个key对应的值进行count计数。</td> 
 </tr> 
 <tr> 
 <td>foreach </td> 
 <td>遍历RDD中的每个元素。 </td> 
 </tr> 
 </tbody> 
</table>

转载于:https://www.cnblogs.com/weiyiming007/p/11101677.html

[image]: /images/20211126/4f97bf1df60e4eee93c9726d71aff8fd.png
[image 1]: https://img2018.cnblogs.com/blog/1190780/201906/1190780-20190628112158224-359482678.png