【Spark】运行原理-CFANZ编程社区

【Spark】运行原理

目录：

一、总体介绍

1.1 Spark集群组成

1.2 逻辑执行图

1.3 物理执行图

二、逻辑执行图

2.1 明确逻辑计划的边界

2.2 RDD的生成

2.3 RDD之间的依赖关系

2.4 RDD之间的依赖关系 — 宽、窄依赖

2.5 窄依赖的类别

三、物理执行图

3.1 简要概述

3.2 Task猜想

3.3 Task划分 — 阶段概念

3.4 stage划分

3.5 数据流动

3.6 查看DAG

四、运行过程

一、总体介绍

1.Spark集群组成:

【Spark】运行原理_返回顶部

案例：

class WordCounts {

  @Test
  def test: Unit ={

    // 1.创建sc
    val conf = new SparkConf().setMaster("local[6]").setAppName("wordCounts")
    val sc   = new SparkContext(conf)

    // 2.读取数据
    val data = sc.textFile("dataset/wordcount.txt")

    // 3.处理数据
    //    3.1 拆分
    val splitRDD = data.flatMap( _.split(" ") )
    //    3.2 赋予词频
    val mapRDD   = splitRDD.map( (_,1) )
    //    3.3 词频统计
    val reduceRDD = mapRDD.reduceByKey( _+_ )
    //    3.4 字符串转换
    val strRDD   = reduceRDD.map( item => s"${item._1},${item._2}" )

    // 4.获取结果
    strRDD.collect().foreach(println(_))

    // 5.关闭sc，释放资源
    sc.stop()
  }

}

【Spark】运行原理_数据_02

2. 逻辑执行图

简单的说，逻辑执行图就是RDD链条，描述的是数据如何处理的步骤。

【Spark】运行原理_spark_03

val conf = new SparkConf().setMaster("local[6]").setAppName("wordCounts")
    val sc   = new SparkContext(conf)

    val data = sc.textFile("hdfs///data/wordcount.txt")
    val splitRDD = data.flatMap( _.split(" ") )
    val mapRDD = splitRDD.map( (_,1) )
    val reduceRDD = mapRDD.reduceByKey( _+_ )
    val strRDD = reduceRDD.map(item => s"${item._1},${item._2}" )

    //strRDD.collect().foreach(println(_))
    println(strRDD.toDebugString)

【Spark】运行原理_返回顶部_04

【Spark】运行原理_spark_05

3. 物理执行图

【Spark】运行原理_返回顶部_06

返回顶部

二、逻辑执行图

2.1 明确逻辑计划的边界

【Spark】运行原理_返回顶部_07

2.2 RDD的生成

1.textFile算子生成的是HadoopRDD

简单查看textFile算子的底层：

【Spark】运行原理_返回顶部_08

【Spark】运行原理_数据_09

2.map算子

【Spark】运行原理_数据_10

【Spark】运行原理_数据_11

3.flatMap算子

【Spark】运行原理_返回顶部_12

上面简单的介绍了一下几个算子的底层，主要目的就是去理解RDD的逻辑计划具体算子的操作流程，进一步明确RDD的逻辑边界。所谓的边界就是指逻辑计划图的始末点。

返回顶部

2.3 RDD之间的依赖关系

what？

【Spark】运行原理_spark_13

RDD之间依赖关系的分水岭 — ShuffledRDD

【Spark】运行原理_spark_14

总结：

【Spark】运行原理_返回顶部_15

返回顶部

2.4 RDD之间的依赖关系 — 宽、窄依赖

1.窄依赖

class narrowDependencyTest {

  val conf = new SparkConf().setMaster("local[6]").setAppName("narrow")
  val sc   = new SparkContext(conf)

  @Test
  def narrowTest: Unit ={

    // 需求：求两个RDD的笛卡尔积
    // 生成RDD
    val rdda=sc.parallelize(Seq(1,2,3,4,5,6))
    val rddb =sc.parallelize(Seq("a","b","c"))

    // 计算
    val rddc= rdda.cartesian(rddb)
    result.collect().foreach(println(_)) // 18个

    sc.stop()
  }
}