Spark教程-低调大师

Spark教程

2017-09-24 683

Spark官方网站

http://spark.apache.org

Spark通用性解释

Spark的特点

官方网站都有。目前支持Java,Scala, Python，R。推荐使用Java和Scala，spark2中对python的支持不够好。

Spark的数据源

HDFS，
HBase，
Cassandra（类似于hbase数据库，国外用的多）
Hive，
Tachyon（基于内存的分布式的文件系统, 阿里出的http://www.alluxio.org/，比较重要）

Spark的四种部署模式

hadoop 模式（spark on yarn）。用yarn来管理spark的资源，也是国内用的最多的模式。
Mesos模式。Mesos一个类似于yarn的资源管理器，国内用的少，国外多。
Standalone模式。上面两种都不是自己管理资源，使用第三方来管理，而standalone模式是spark自己来管理资源，多内用的比较多
部署到云端

验证集群是否好用

验证spark集群是否可用

./bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --num-executors 1 --driver-memory 500m --executor-memory 500m --executor-cores 1 lib/spark-examples-1.5.2-hadoop2.5*.jar 10

验证结果如下：
Pi is roughly 3.1411


用scala编写完简单的运算后，可以在localhost:4040上查看

运行spark有两种：

本地运行

bin/spark-submit --master local --class org.apache.spark.examples.SparkPi lib/spark-examples-1.5.2-hadoop2.6.0-cdh5.4.4.jar 5

本地运行会直接输出结果

分布式运行

bin/spark-submit --master yarn-cluster --class org.apache.spark.examples.SparkPi lib/spark-examples-1.5.2-hadoop2.6.0-cdh5.4.4.jar 5

而分布式运行的结果在

cd $HADOOP_HOME/logs/userlogs/

scala 运行例子

val lines = sc.textFile("hdfs://master:9000/input0917/qqFriend.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

输出：
totalLength: Int = 188

RDD

Spark架构

Spark安装前准备

安装好JAVA
安装好Hadoop
安装好scala
并配置好上述的的环境变量

Spark任务提交

http://spark.apache.org/docs/2.1.0/submitting-applications.html

SparkHistoryServer配置

关于sparkshell的使用

http://blog.csdn.net/sunflower_cao/article/details/26708797

创见RDD文件

第一种方式是通过读取本地或者hdfs上的文件创建RDD

sc.textFile("hdfs://")

第二种方式是通过并行化的方式创建RDD. 其实就是通过我们自己取模拟数据

val str=Array("you jump","I jump")
val list = Array(1,2,3,4,5,6)
val listadd = sc.parallelize(list) 可以看到返回值就是RDD，当然可以调用RDD中的函数，比如reduce算子等
listadd.reduce(_+_)

大多数方式是使用第一种

transformation 和action 原理

Spark支持两种RDD操作: transformation and action.
transformation 会针对已有的RDD创建一个新的RDD，而action则主要对RDD进行最后的操作。transformation只是记录了对RDD的操作，并不会触发spark程序的执行，只有当transform之后接着一个action操作，那么所有的transformation才会执行。比如

val file=sc.textFile("hdfs://hadoop1:9000/hello.txt").flatMap(line => line.split("\t"))
回车之后并没有触发spark的执行，因为flatMap等属于transformation操作
等到file.collect()后会看到spark的执行，collect是action操作

具体可以参考编程指南：http://spark.apache.org/docs/2.1.0/programming-guide.html

那为什么要有transformation和action呢？Spark可以通过这种lazy属性，来进行底层的spark应用程序的优化，避免过多的中间结果。

Spark算子

http://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations

练习如下，注意eclipse的使用。


 - 将鼠标放在方法上，ctrl+1 来选择返回值类型
 - 鼠标选中区域, ctrl+shift+/ 来进行多行注释
 - 当实现某个方法，查看错误信息，提示添加内部方法，添加即可

map和flatmap的区别：

map与mappartitions的区别：

map迭代的是RDD中的每一个元素，而mappartiontions迭代的一个分区。如果在映射过程中频繁创建额外的对象，那么mappartitions比map高效。但也需要注意内存，因为内存可能不够用，比分区小。

比如在迭代元素的时候要将10000条结果存到mysql数据库，如果用map需要创建10000个额外的connection，用map partitions可能只需要创建10个, 对每个分区建立连接。

下面是不同算子做实验的代码示例：

repartition coalesce 重新进行分区宽以来窄依赖 shuffle
那什么时候用repartition呢？比如： filter之后，partition的量会减少，比如之前是100 partitions，对应的需要起100 个tasks，现在filter后变成了50个，此时要是100个tasks就浪费了，可以repartition到50。repartition分区会进行shuffle操作

package com.ecaoyng.spark;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.cassandra.cli.CliParser.newColumnFamily_return;
import org.apache.cassandra.thrift.Cassandra.system_add_column_family_args;
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.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import org.stringtemplate.v4.compiler.STParser.mapExpr_return;

import scala.Tuple2;

public class Trans {

    /**
     * repartition will do shuffle 
     */

    public static void repartition(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("repartition");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4,5,6);
        JavaRDD<Integer> list1RDD = sc.parallelize(list1);
        JavaRDD<Integer> repartition = list1RDD.repartition(2);
        repartition.foreach(new VoidFunction<Integer>() {

            public void call(Integer arg0) throws Exception {
                System.out.println(arg0);
            }

        });

    }




    /**
     * output: 
     * Hello1
Hello2
Hello3

Hello4
Hello5
Hello6

     */


    public static void mapPartitions(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("mapParititons");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4,5,6);
        JavaRDD<Integer> list1RDD = sc.parallelize(list1, 2);
        /**
         * return value is string
         */
        JavaRDD<String> mapPartitions = list1RDD.mapPartitions(new FlatMapFunction<Iterator<Integer>, String>() {

            /*
             * process each partition each time
             * partition1: 123
             * partition2: 456
             * */

            public Iterable<String> call(Iterator<Integer> t)
                    throws Exception {
                ArrayList<String> arrayList = new ArrayList<String>();
                while (t.hasNext()) {
                    Integer i = (Integer) t.next();
                    arrayList.add("Hello" + i);

                }   

                return arrayList;
            }

        });
        mapPartitions.foreach(new VoidFunction<String>() {

            public void call(String arg0) throws Exception {
                System.out.println(arg0);
            }


        });
    }


    /*
     * di ka er ji
     * output:1 a
1 b
1 c
2 a
2 b
2 c
3 a
3 b
3 c
    */
    public static void cartesian(){

        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("cartesian");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3);
        List<String> list2 = Arrays.asList("a","b","c");
        JavaRDD<Integer> list1RDD = sc.parallelize(list1);
        JavaRDD<String> list2RDD = sc.parallelize(list2);
        JavaPairRDD<Integer, String> cartesian = list1RDD.cartesian(list2RDD);
        cartesian.foreach(new VoidFunction<Tuple2<Integer,String>>() {

            public void call(Tuple2<Integer, String> t) throws Exception {
                System.out.println(t._1 + " " + t._2);
            }
        });


    }


/*  qu chong
 * output:4
1
6
3
5
2
    */

    public static void distinct(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("distinct");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4,4,5,5,6,6);
        JavaRDD<Integer> listRDD = sc.parallelize(list1);
        JavaRDD<Integer> distinct = listRDD.distinct();
        distinct.foreach(new VoidFunction<Integer>() {

            public void call(Integer t) throws Exception {
                System.out.println(t);
            }

        });


    }


    /*
     * jiao ji
     * output: 4
3
    */
    public static void interSection() {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("union");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4);
        List<Integer> list2 = Arrays.asList(3,4,5,6);
        JavaRDD<Integer> list1RDD = sc.parallelize(list1);
        JavaRDD<Integer> list2RDD = sc.parallelize(list2);
        JavaRDD<Integer> intersection = list1RDD.intersection(list2RDD);

        intersection.foreach(new VoidFunction<Integer>() {

            public void call(Integer t) throws Exception {
                System.out.println(t);
            }
        });

    }


    /*
     * bingji, bu qu chong
    */
    public static void union() {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("union");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4);
        List<Integer> list2 = Arrays.asList(3,4,5,6);
        JavaRDD<Integer> list1RDD = sc.parallelize(list1);
        JavaRDD<Integer> list2RDD = sc.parallelize(list2);
        JavaRDD<Integer> union = list1RDD.union(list2RDD);
        union.foreach(new VoidFunction<Integer>() {

            public void call(Integer t) throws Exception {
                System.out.println(t);
            }
        });

    }




    /*
     * ID: 1
Name: [Tom, Tom]
Score: [30, 70]
ID: 3
Name: [Chirs]
Score: [60, 90]
ID: 2
Name: [Jerry]
Score: [40, 60]
     * 
    */
    public static void cogroup() {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("cogroup");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> stu = Arrays.asList(
                    new Tuple2<Integer, String>(1, "Tom"),
                    new Tuple2<Integer, String>(2, "Jerry"),
                    new Tuple2<Integer, String>(3, "Chirs"),
                    new Tuple2<Integer, String>(1, "Tom")
                    );

        List<Tuple2<Integer, Integer>> score = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 30),
                new Tuple2<Integer, Integer>(2, 40),
                new Tuple2<Integer, Integer>(3, 60),
                new Tuple2<Integer, Integer>(1, 70),
                new Tuple2<Integer, Integer>(2, 60),
                new Tuple2<Integer, Integer>(3, 90)
                );

        JavaPairRDD<Integer, String> stuRDD = sc.parallelizePairs(stu);
        JavaPairRDD<Integer, Integer> scoreRDD = sc.parallelizePairs(score);
//      <2,tuple<Jerry,{40,60}>>
        JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> cogroupRDD = stuRDD.cogroup(scoreRDD);
        cogroupRDD.foreach(new VoidFunction<Tuple2<Integer,Tuple2<Iterable<String>,Iterable<Integer>>>>() {

            public void call(
                    Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> t)
                    throws Exception {
                System.out.println("ID: " + t._1);
                System.out.println("Name: " + t._2._1);
                System.out.println("Score: " + t._2._2);
            }
        });
    }



    /*
     * output:
ID: 1
Name: Tom
Score: 30
==================>
ID: 3
Name: Chirs
Score: 60
==================>
ID: 2
Name: Jerry
Score: 40
==================>
    */

    public static void join(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("join");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> stu = Arrays.asList(
                    new Tuple2<Integer, String>(1, "Tom"),
                    new Tuple2<Integer, String>(2, "Jerry"),
                    new Tuple2<Integer, String>(3, "Chirs")
                    );

        List<Tuple2<Integer, Integer>> score = Arrays.asList(
                new Tuple2<Integer, Integer>(1, 30),
                new Tuple2<Integer, Integer>(2, 40),
                new Tuple2<Integer, Integer>(3, 60)
                );

        JavaPairRDD<Integer, String> stuRDD = sc.parallelizePairs(stu);
        JavaPairRDD<Integer, Integer> scoreRDD = sc.parallelizePairs(score);
//      Integer:id, Tuple2<String, Integer>: name,score
        JavaPairRDD<Integer, Tuple2<String, Integer>> joined = stuRDD.join(scoreRDD);
        joined.foreach(new VoidFunction<Tuple2<Integer,Tuple2<String,Integer>>>() {

            public void call(Tuple2<Integer, Tuple2<String, Integer>> t)
                    throws Exception {
                System.out.println("ID: " + t._1);
                System.out.println("Name: " + t._2._1);
                System.out.println("Score: " + t._2._2);
                System.out.println("==================>");
            }

        });




    }




    public static void sortByKey(){

    }

    /* calculate sum by each key
     * output:
PBCS:95
TDCS:120
    */
    public static void reduceByKey(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("reduceByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<String, Integer>> asList = Arrays.asList(
                    new Tuple2<String, Integer>("TDCS", 30),
                    new Tuple2<String, Integer>("TDCS", 40),
                    new Tuple2<String, Integer>("TDCS", 50),
                    new Tuple2<String, Integer>("PBCS", 40),
                    new Tuple2<String, Integer>("PBCS", 55)             
                    );

        JavaPairRDD<String, Integer> parallelizePairsRDD = sc.parallelizePairs(asList);
        /*
         * input: Integer, Integer
         * output: sum:Integer
         * scala.reduceByKey(_+_)
        */
        parallelizePairsRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {

            public Integer call(Integer v1, Integer v2) throws Exception {

                return v1 + v2;
            }
        }).foreach(new VoidFunction<Tuple2<String,Integer>>() {

            public void call(Tuple2<String, Integer> sum) throws Exception {
                System.out.println(sum._1 + ":" + sum._2);
            }
        });

    }



    /*
     * src: key-value   
     * Tuple2 in scala is similar with map in Java  
     * output:
PBCS
Terry
Shawn
======================================>
TDCS
Tom
Jerry
Chris
     * 
    */
    public static void groupByKey(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("groupByKey");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<String, String>> list = Arrays.asList(
                new Tuple2<String, String>("TDCS", "Tom"),
                new Tuple2<String, String>("TDCS", "Jerry"),
                new Tuple2<String, String>("TDCS", "Chris"),
                new Tuple2<String, String>("PBCS", "Terry"),
                new Tuple2<String, String>("PBCS", "Shawn")             
                );
//      due to key-value, it should be use parallelizePairs 
        JavaPairRDD<String, String> parallelizePairsRDD = sc.parallelizePairs(list);
        JavaPairRDD<String, Iterable<String>> groupByKey = parallelizePairsRDD.groupByKey();
        groupByKey.foreach(new VoidFunction<Tuple2<String,Iterable<String>>>() {

            public void call(Tuple2<String, Iterable<String>> t)
                    throws Exception {
                System.out.println(t._1);
                Iterator<String> iterator = t._2.iterator();
                while (iterator.hasNext()) {
                    System.out.println(iterator.next());

                }
                System.out.println("======================================>");

            }

        });
    }


    /*
     * flatmap: {You jump I jump}
     * split each word into a sigle line 
    */
    public static void flatMap(){
        SparkConf conf = new SparkConf();
//      if no set on conf.setMaster, the default setting will use distributed mode. 
        conf.setMaster("local");
        conf.setAppName("flatmap");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list = Arrays.asList("you jump","I jump");
        JavaRDD<String> listRDD = sc.parallelize(list);
//      U refers to the return value of method: FlatMapFunction
        JavaRDD<String> flatMap = listRDD.flatMap(new FlatMapFunction<String, String>() {

            public Iterable<String> call(String t) throws Exception {
                return Arrays.asList(t.split(" "));
            }
        });

        flatMap.foreach(new VoidFunction<String>() {

            public void call(String line) throws Exception {
                System.out.println(line);
            }
        });




    }


    /*
     * Map() test: say hello to each items within a array
    */
    public static void map() {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("map");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list =  Arrays.asList("Tom","Jerry","USA","China");
        JavaRDD<String> listRDD = sc.parallelize(list);
//      R refers to the retuen value of new Function
        JavaRDD<String> map = listRDD.map(new Function<String, String>() {

            public String call(String str) throws Exception {
                return "hello " + str;
            }
        });

//      action opts
        map.foreach(new VoidFunction<String>() {

            public void call(String str) throws Exception {
                System.out.println(str);
            }
        });


    }

    /*
     * fliter(): get odds from dataset
    */
    public static void fliter(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("map");
        JavaSparkContext sc_fliter = new JavaSparkContext(conf);
        List<Integer> fliter_list = Arrays.asList(1,2,3,4,5,6,7,8,9);
        JavaRDD<Integer> fliterRDD = sc_fliter.parallelize(fliter_list);
        JavaRDD<Integer> filter = fliterRDD.filter(new Function<Integer, Boolean>() {

            public Boolean call(Integer num) throws Exception {
                return num %2 ==0;
            }
        });
        filter.foreach(new VoidFunction<Integer>() {

            public void call(Integer num) throws Exception {
                System.out.println(num);
            }

        });









    }




    public static void main(String[] args) {
//      map();
//      fliter();
//      flatMap();
//      groupByKey();
//      reduceByKey();
//      join();
//      cogroup();
//      union();
//      interSection();
//      distinct();
//      cartesian();
//      mapPartitions();
        repartition();


    }
}

RDD持久化（RDD Persistence）

http://spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence

当你缓存了一个RDD，每个节点都缓存了RDD的所有分区，这样就可以在内存中进行计算，速度更快(提高10倍)。
可以对希望缓存的RDD进行cache或者persist方法进行标记。她通过动作操作第一次在Rdd上进行计算后，它会被缓存在节点的内存中。Spark的缓存具有容错性，如果RDD的某个分区丢失，它会自动使用最初创建RDD的转换操作进行重新计算。

建议：

优先选择MEMORY_ONLY（=cache），纯内存计算，速度快
MEMORY_ONLY 缓存不了所有的数据，MEMORY_ONLY_SER 将数据序列化后进行存储，需要耗费一些CPU资源，并需要反序列化。
可以选在带2的方式。恢复的时候可以使用备份，不需要重新计算
能不是用disk的就不使用disk，有时候从磁盘读还不如重新计算

之外，OFF_HEAP (experimental) 是将RDD存到tachyon上。

那什么是tachyon

一个开源的基于jvm的内存分布式文件系统，介于计算层和存储层之间，简单的理解为存储层在内存内的缓存系统。

为何会出现tachyon？以内存替换磁盘，就能明显的减少延时，所以涌现出很多基于内存的计算工具，比如spark计算框架。

spark运行在jvm中，spark的任务会将数据存入jvm的堆中，随着计算的迭代，jvm堆中存放的数据量迅速增大，对于spark而言，spark的计算引擎和存储引擎处在同一个jvm中，所以会有重复的gc方面的开销，增大了系统的延时
1. 当jvm崩溃时，缓存在jvm堆中的数据也会消失，这个时候spark不得不根据rdd的血缘关系重新计算数据。
2. 如果spark需要和其他的计算框架，比如mapreduce，此时就需要通过第三方共享，比如hdfs。需要额外的开销，比如磁盘的IO开销

因基于内存的分布式计算框架有以上的问题，那么就促使了内存分布式文件系统的诞生，比如tachyon。

共享变量（广播变量，累加变量）

广播变量

举例：之前 val a的时候需要将a拷贝好多份到每个task上，如果a大道1个G，就浪费好多空间。现在有了broadcast var 就将变量拷到

累加变量

按照之前的习惯写sum会出错，一种新的声明方式

Spark on Yarn

只需要在spark下的conf目录下的spark-env.sh 中配置Hadoop_conf_dir（hadoop的位置）即可。让spark通过hadoop的配置文件找到yarn.

之前用standalone模式需要启动master和worker，现在用yarn模式不需要启动master和worker，只需要启动hdfs和yarn即可

start-dfs.sh, jps -> 3个服务
start-yarn.sh

之后spark的开发和之前一样，只需要在提交代码的时候指定运行的模式
具体请参考官方文档：
http://spark.apache.org/docs/latest/submitting-applications.html

# Run on a YARN cluster
export HADOOP_CONF_DIR=XXX
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn \
  --deploy-mode cluster \  # can be client for client mode
  --executor-memory 20G \
  --num-executors 50 \
  /path/to/examples.jar \
  1000

yarn提交也分为client和cluster模式

cluster

注意executor会向driver（application master）

client：

建议：

调试的时候使用client模式。使用client模式的时候打印出来的信息非常详细，有利于我们调试程序。
调试完成以后，建议使用cluster模式提交任务。

举例，先在本地运行如下代码，成功在本地生成union的并集：

package com.ecaoyng.spark;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

public class SaveAsText {

    public static void Save(){
        SparkConf conf = new SparkConf();
        conf.setMaster("local[3]"); // 因为是本地运行
        conf.setAppName("SaveAsTextFile");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list1 = Arrays.asList(1,2,3,4,5);
        List<Integer> list2 = Arrays.asList(4,5,6,7,8);
        JavaRDD<Integer> list1RDD = sc.parallelize(list1);
        JavaRDD<Integer> list2RDD = sc.parallelize(list2);
        JavaRDD<Integer> unionRDD = list1RDD.union(list2RDD);
        unionRDD.repartition(1).saveAsTextFile("/home/zkpk/union");
        # 注意是在本地磁盘的目录下生成union文件夹

    }


    public static void main(String[] args) {
        Save();
    }


}

测试通过后，因为要放到cluster上运行，所以将conf.setMaster注释掉。之后用eclipse将jar包导出
下面是提交的代码：

/home/zkpk/spark-1.5.2-bin-2.5.2/bin/spark-submit \
  --class com.ecaoyng.spark.SaveAsText \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 100M \
  --num-executors 1 \
  /home/zkpk/output/SaveUnion.jar \

注意里面不要出现注释，否则可能会出错。如果发现参数错误可以spark-submit –help 查看哪些是可以运行在cluster上的参数。输出的结果放在hdfs上。当然事先需要设置export SPARK_CONF_DIR=/home/zkpk/hadoop-2.5.2 或者在spark的配置中的spark-env.sh中配置
也可以在web中查看http://master:18088/ 提交的进度

宽依赖和窄依赖

窄依赖，父RDD的每个分区只被一个子RDD的分区依赖。
宽依赖，父RDD的分区被多个子RDD的分区所依赖。

shuffle的原理

本例中前面三个RDD都不涉及到其他的节点，但是真实环境是需要其他节点的参与的。shuffle阶段需要大量的磁盘IO，序列化与反序列化，网络数据的传输，所以spark很大的性能损耗都在shuffle上，所以有必要进行调优。

shuffle的发展经历了三个阶段：1.2，1.3, 1.5之后。分别取名：

未经优化的HashshuffleManager
优化后的HashshuffleManager
SortShuffleManager
- 普通机制
- bypass机制

[未经优化的HashshuffleManager]

每一个map task生成的buffer和file和 reduce的数量有关，可以联想MR中的partition，可以看到生成的碎文件太多，产生的中间文件数量如图

[经过优化的HashshuffleManager]

[现在在使用的是sort-based shuffle manage ]

shuffle 配置参数地址：http://spark.apache.org/docs/2.0.0/configuration.html

正常情况使用的是普通运行机制，但是当shuffle read task小于200（默认值）时就启用bypass机制

因为合并成了一个磁盘文件，所以中间文件将被删除

bypass模式

未完待续

微信关注我们

原文链接：https://yq.aliyun.com/articles/350211

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

全球计算奥运会阿里云破四个世界纪录

本文讲的是全球计算奥运会阿里云破四个世界纪录【IT168云计算】10月28日，Sort Benchmark在官方网站公布了2015年排序竞赛的最终成绩。阿里云用不到7分钟(377秒)就完成了100TB的数据排序，打破了Apache Spark的纪录23.4分钟。Sort Benchmark是全球科技公司的计算奥运会，更早之前Hadoop的记录是72分钟。 Sort Benchmark的排序竞赛包含4项比赛，其中GraySort和MinuteSort主要评测计算系统的规模和效率，是4项比赛中含金量最高的两项。每一项比赛又分通用目的排序(Daytona)和专用目的排序(Indy)两个类别。阿里云飞天分布式计算系统，在这两项重量级比赛中打破了全部4项世界纪录。 GraySort评测的是超大规模数据量排序的速度。飞天用了377秒对100TB数据完成排序，也就是说1分钟能够完成15.9TB的数据排序。去年Apache Spark成绩为4.35TB/分钟。MinuteSort评测的是系统在1分钟内能够完成排序的数据量。飞天用1分钟完成了对7.7TB数据的排序，去年冠军三星为3.7TB。阿里云首...

2017-09-25

536

1：首先搞好实体类对象： write 是把每个对象序列化到输出流，readFields是把输入流字节反序列化，实现WritableComparable，Java值对象的比较：一般需要重写toString(),hashCode(),equals()方法 1 package com.areapartition; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 7 import org.apache.hadoop.io.Writable; 8 import org.apache.hadoop.io.WritableComparable; 9 10 /*** 11 * 12 * @author Administrator 13 * 1:write 是把每个对象序列化到输出流 14 * 2:readFields是把输入流字节反序列化 15 * 3:实现WritableComparable 16 * Java值对象的比较：一般需要重写toString(),hashC...

2017-09-25

473

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。

Spark教程

Spark官方网站

Spark通用性解释

Spark的特点

Spark的数据源

Spark的四种部署模式

验证集群是否好用

RDD

Spark架构

Spark安装前准备

Spark任务提交

SparkHistoryServer配置

关于sparkshell的使用

创见RDD文件

transformation 和action 原理

Spark算子

RDD持久化（RDD Persistence）

那什么是tachyon

共享变量（广播变量，累加变量）

Spark on Yarn

宽依赖和窄依赖

shuffle的原理

全球计算奥运会阿里云破四个世界纪录

一脸懵逼学习Hadoop中的MapReduce程序中自定义分组的实现

相关文章

发表评论

资源下载

腾讯云软件源

Spring

Rocky Linux

Sublime Text

欢迎您！

Spark教程

Spark官方网站

Spark通用性解释

Spark的特点

Spark的数据源

Spark的四种部署模式

验证集群是否好用

RDD

Spark架构

Spark安装前准备

Spark任务提交

SparkHistoryServer配置

关于sparkshell的使用

创见RDD文件

transformation 和action 原理

Spark算子

RDD持久化（RDD Persistence）

那什么是tachyon

共享变量（广播变量，累加变量）

Spark on Yarn

宽依赖和窄依赖

shuffle的原理

全球计算奥运会 阿里云破四个世界纪录

一脸懵逼学习Hadoop中的MapReduce程序中自定义分组的实现

相关文章

发表评论

资源下载

腾讯云软件源

Spring

Rocky Linux

Sublime Text

欢迎您！

全球计算奥运会阿里云破四个世界纪录