प्रौद्योगिकी साझेदारी

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

स्पार्क मूलभूत बातें

संक्षिप्त वर्णन

1. स्पार्क जॉब निष्पादनस्य विशेषताः : १.

(1) यदा एव एक्शन-सञ्चालकस्य सम्मुखीभवति तदा एव सम्पूर्णं स्पार्क-कार्यं निष्पादनार्थं प्रवर्तते ।

(२) अनेकवारं सम्मुखीकृतः, अनेकवारं निष्पादितः

2. RDD: लोचदारवितरितदत्तांशसमूहः

लचीलापनम् : दत्तांशस्य परिमाणं बृहत् लघु वा भवितुम् अर्हति

RDD पात्रस्य सदृशं भवति, परन्तु एतत् दत्तांशं न संगृह्णाति, अपितु गणनातर्कं संगृह्णाति ।

यदा कश्चन क्रिया-सञ्चालकः सम्मुखीभवति तदा प्रथम-आरडीडी-तः आरभ्य सम्पूर्णं स्पार्क-कार्यं निष्पादयितुं प्रवर्तते ।

दत्तांशः केवलं आरडीडी-मध्ये प्रवाहितः सम्बन्धः अस्ति, सः न संगृहीतः भविष्यति ।

प्रवाहितस्य दत्तांशस्य परिमाणं बृहत् लघु वा भवितुम् अर्हति अतः तत् लोचना कथ्यते

वितरितः : १.

स्पार्कस्य मूलतः HDFS तः आँकडानां पठनस्य आवश्यकता वर्तते HDFS वितरितं भवति तथा च भविष्ये आँकडाखण्डाः भिन्न-भिन्न-दत्तांश-नोड्-मध्ये भवितुम् अर्हन्ति ।

RDD मध्ये प्रवाहितः दत्तांशः भिन्न-भिन्न-दत्तांश-नोड्-मध्ये ब्लॉक-दत्तांशतः आगन्तुं शक्नोति ।

दत्तांशसमूहः : १.

गणनाप्रवाहप्रक्रियायाः समये RDD संक्षेपेण पात्ररूपेण गणयितुं शक्यते पात्रे दत्तांशः अस्ति, यः पूर्वनिर्धारितरूपेण स्मृतौ न संगृह्यते ।

पश्चात् RDD इत्यस्य दत्तांशं डिस्कमध्ये संग्रहीतुं मार्गः भविष्यति ।

आरडीडी इत्यस्य पञ्च प्रमुखलक्षणाः : (साक्षात्कारेषु अवश्यं पृच्छन्तु!)

1. RDD विभाजनस्य श्रृङ्खलायाम् अस्ति

1) सञ्चिकां पठन् minPartitions पैरामीटर् केवलं न्यूनतमं विभाजनसङ्ख्यां निर्धारयितुं शक्नोति सञ्चिकां पठित्वा RDD विभाजनस्य वास्तविकसङ्ख्या स्वयं दत्तांशसामग्रीभिः तथा च क्लस्टरस्य वितरणेन निर्धारिता भवति ।

२) यदि minPartitions सेट् कृते संख्या खण्डसङ्ख्यायाः अपेक्षया न्यूना भवति तर्हि विभाजनस्य संख्या वस्तुतः खण्डसङ्ख्यायाः आधारेण निर्धारिता भवति ।

3) shuffle जनयति इति ऑपरेटरं आह्वयति सति, भवान् numPartitions (उदाहरणार्थं: groupby()) इत्यत्र पारयितुं शक्नोति, यत् वास्तवतः RDD इत्यस्य विभाजनस्य संख्यां परिवर्तयितुं शक्नोति यत् भवान् कति विभाजनं सेट् करोति तत् निर्धारयिष्यति यत् अन्तिमः RDD कियन्तः विभाजनं करिष्यति अस्ति।

४) सञ्चिका HDFS मध्ये खण्डरूपेण संगृहीता भविष्यति यदि सञ्चिका 128M इत्यस्य पूर्वनिर्धारितमूल्यं न प्राप्नोति तर्हि सा अपि खण्डे संगृहीता भविष्यति ।

प्रारम्भे RDD मध्ये विभाजनस्य संख्या दत्तांशं पठन्तः खण्डानां संख्यायाः आधारेण निर्धारिता भवति ।

उत्तर-आरडीडी-मध्ये विभाजनदत्तांशः, केवी-कार्यं विहाय, पूर्व-आरडीडी-मध्ये विभाजन-दत्तांशस्य तार्किक-संसाधनस्य परिणामेण सह सङ्गच्छते । पूर्वनिर्धारितरूपेण, यदि अनन्तरं विभाजनं न संसाधितं भवति तर्हि सर्वेषां अनन्तरं RDD-विभाजनस्य संख्या प्रथमस्य RDD इत्यस्य उपरि निर्भरं भवति ।

अन्तिम RDD मध्ये अनेकाः विभाजनानि सन्ति, भविष्ये च HDFS मध्ये अनेकाः परिणामसञ्चिकाः द्रक्ष्यन्ति (HDFS -> RDD -> HDFS) ।

2. संचालकः प्रत्येकं विभाजने कार्यं करोति (प्रत्येकं विभाजनं संसाधितं भविष्यति)

3. आरडीडी आरडीडी च मध्ये केचन आश्रयाः सन्ति

१) संकीर्णतया एक-एक-सम्बन्धे निर्भरं भवति यत्र पूर्व-आरडीडी-मध्ये निश्चित-विभाजनात् दत्तांशः केवलं अग्रिम-आरडीडी-मध्ये अद्वितीय-विभाजनं प्रति गमिष्यति (अथवा बहुविध-पूर्व-विभाजनं अग्रिम-विभाजनं प्रति गन्तुं शक्नोति)

2) पूर्वस्मिन् RDD मध्ये कस्यचित् विभाजनस्य व्यापकनिर्भरतायाः सह एक-बहु-सम्बन्धः परवर्ती RDD मध्ये भिन्न-भिन्न-विभाजनेषु प्रविशति भवान् अपि shuffle उत्पन्नः अस्ति वा इति जाँचयितुम् अर्हति ।

3) सम्पूर्णं स्पार्कजॉबं विस्तृतनिर्भरतायाः संख्यायाः द्वारा अनेकचरणयोः विभक्तं भविष्यति, Num(stage) = Num(wide dependencies) + 1

४) यदा शफलं जनयति तस्य संचालकस्य सम्मुखीभवति तदा तस्य समावेशः भवतिपूर्वस्य RDD तः डिस्कं प्रति दत्तांशं लिखित्वा डिस्कतः अग्रिम RDD यावत् दत्तांशं पठितुं च घटना ।

नोट्: यदा प्रथमवारं निष्पादनं प्रवर्तते तदा डिस्कमध्ये कोऽपि दत्तांशः नास्ति, अतः प्रथमजनित RDD तः निष्पादनं आरभ्यते ।

यदा समानं निष्पादनं पुनः पुनः प्रवर्तते, तस्यैव DAG निर्देशितस्य अचक्रीयग्राफस्य कृते, निष्पादनं शफलस्य अनन्तरं RDD तः प्रत्यक्षतया आरभ्यते (पूर्वस्य RDD तः डिस्कं प्रति दत्तांशलेखनस्य प्रक्रियां त्यक्त्वा), तथा च तत् प्रत्यक्षतया डिस्कतः पठितुं शक्यते । दत्तांश।

५) **एकस्मिन् चरणे, २.आरडीडी अस्तिअनेकविभाजनेन सह अनेकाः समानान्तरकार्यं भविष्यति ।

4. kv संचालकः केवलं kv इत्यस्य RDD इत्यत्र एव कार्यं कर्तुं शक्नोति।

5. स्पार्क् इष्टतमं कार्यगणनाविधिं प्रदास्यति, केवलं गणनां चालयति परन्तु दत्तांशं न।

स्पार्कस्य एकः डिजाइनसिद्धान्तः अस्तिदत्तांशस्थानीयीकरणम्(Data Locality), अर्थात् यत्र दत्तांशः स्थितः तस्मिन् नोड् मध्ये कम्प्यूटिंग् कार्याणि निष्पादयितुं प्रयतन्ते, येन दत्तांशस्य संजालसंचरणस्य उपरिभारः न्यूनीकरोति

स्पार्क उदाहरणम् : शब्दगणना

object WordCount2 {
  def main(args: Array[String]): Unit = {
    //创建spark配置文件对象
    val conf: SparkConf = new SparkConf()
    //设置运行模式
    //如果是本地local模式运行的话，需要设置setMaster
    //将来如果是集群进行，将这句话注释即可
    conf.setMaster("local")
    //设置spark作业的名字
    conf.setAppName("wordcount")

    //创建spark core上下文环境对象
    val sc: SparkContext = new SparkContext(conf)
    //===================================================================================

    //读取文件,每次读取一行
    //RDD是spark core中的核心数据结构，将来运行的时候，数据会在RDD之间流动，默认基于内存计算
    val linesRDD: RDD[String] = sc.textFile("spark/data/wcs/*")
    //    println(s"linesRDD的分区数：${linesRDD.getNumPartitions}")

    //一行数据根据分隔符分割
    val wordRDD: RDD[String] = linesRDD.flatMap(_.split("\|"))
    //    println(s"wordRDD的分区数：${wordRDD.getNumPartitions}")


    //将每一个单词组成(word,1)
    val kvRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
        println(s"kvRDD的分区数：${kvRDD.getNumPartitions}")

    //根据键进行分组，并设置分区数为 5
    val kvRDD2: RDD[(String, Iterable[(String, Int)])] = kvRDD.groupBy(_._1,numPartitions = 5)
        println(s"kvRDD2的分区数：${kvRDD2.getNumPartitions}")

    val resRDD: RDD[(String, Int)] = kvRDD2.map((e: (String, Iterable[(String, Int)])) => (e._1, e._2.size))
        println(s"resRDD的分区数：${resRDD.getNumPartitions}")

    //打印
    resRDD2.foreach(println)

    //指定的是所要写入数据的文件夹的路径
    //spark如果是local本地运行的话，会将本地文件系统看作一个hdfs文件系统
    resRDD.saveAsTextFile("spark/data/outdata1")

  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

स्पार्क् मध्ये RDD द्वारा आहूतानि कार्याणि ऑपरेटर् इति उच्यन्ते

संचालकाः द्वयोः वर्गयोः विभक्ताः सन्ति- १.

1. रूपान्तरण संचालकः (RDD -> RDD, प्रसंस्करण तर्कः) .

2. क्रिया संचालकः (कार्यनिष्पादनं प्रेरयति) .

1. रूपान्तरणसञ्चालकः

१)नक्शा

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1Map {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("map算子演示")
    val sc: SparkContext = new SparkContext(conf)

    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //map操作算子：将rdd中的数据依次取出，传递给后面函数逻辑，将计算后的数据返回到新的rdd中
    //将rdd中的数据依次取出，处理完的数据返回下一个rdd直接继续执行后续的逻辑
    val rdd2: RDD[(String,String,String,String,String)] = lineRDD.map((line: String) => {
      println("==============处理后的数据========================")
      val array1: Array[String] = line.split(",")
      (array1(0),array1(1),array1(2),array1(3),array1(4))
    })

    //foreach是一个行动算子，遇到行动算子，触发作业执行
    /**
     * 转换操作（转换算子中定义了操作逻辑）仅仅是定义了数据应该如何被转换，而不会立即执行。
     * 只有当需要计算一个结果时（即调用行动算子时），才会执行。
     * 打印结果：
     * ==============处理后的数据========================
     * (1500100001,施笑槐,22,女,文科六班)
     * ==============处理后的数据========================
     * (1500100002,吕金鹏,24,男,文科六班)
     *每次调用行动算子（foreach）打印一条数据，都会是整个RDD重新执行一次（所有RDD的执行关系是一个有向无环图）
     */
    rdd2.foreach(println)

  }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

२)छिद्रम्

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo2Filter {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("map算子演示")
    val sc: SparkContext = new SparkContext(conf)

    //===============================================================
    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //需求：过滤出所有的男生
    //filter转换算子：将rdd中的数据依次取出，传递给后面的函数，跟map一样，也是依次传递一条
    // 若不匹配，则无数据在RDD间流动，在下面执行.foreach(println)时也无数据进行打印，
    // 但是判断中的println()属于scala,并不受影响
    val genderRDD: RDD[String] = lineRDD.filter((line: String) => {
      var b: Boolean = false
      if ("女".equals(line.split(",")(3))) {
        println("============这是女生==================")
      } else {
        println("============这是男生==================")
        b = "男".equals(line.split(",")(3))
      }
      b
    })

    genderRDD.foreach(println)

  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

३)सपाटमपः

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo3FlatMap {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("map算子演示")
    val sc: SparkContext = new SparkContext(conf)

    //===============================================================
    val lineRDD: RDD[String] = sc.textFile("spark/data/wcs/words.txt")

    /**
     * flatMap: 将rdd中的每一条数据传递给后面的函数，最终将返回的数组或者是序列进行扁平化，返回给新的集合
     * 由于flatMap会“扁平化”结果，因此words RDD将包含所有分割后的单词，而不是单词数组(返回一个元素为单个单词的集合)。
     * 打印结果：
     * ===============一条数据====================
     * hello
     * world
     * ===============一条数据====================
     * java
     * hadoop
     * linux
     */
    val rdd1: RDD[String] = lineRDD.flatMap((line:String)=>{
      println("===============一条数据====================")
      line.split("\|")
    })

    rdd1.foreach(println)


  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

४)नमूना

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo4Sample {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("map算子演示")
    val sc: SparkContext = new SparkContext(conf)

    //===============================================================
    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    /**
     * sample抽样，1000条数据，抽0.1比例，结果的数量在100左右
     * 这个函数主要在机器学习的时候会用到
     * withReplacement ：
     * 为True时，抽样结果中可能会包含重复的元素。
     * 为False时，抽样结果中不会包含重复的元素。
     * fraction：这是一个浮点数（Double），指定了抽样的比例，取值范围在[0, 1]之间。
     */
    val rdd1: RDD[String] = lineRDD.sample(withReplacement = false, fraction = 0.1)

    rdd1.foreach(println)
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

५)समूहBy

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo5GroupBy {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("groupBy")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //求每个班级的平均年龄
    val arrayRDD: RDD[Array[String]] = linesRDD.map((line: String) => line.split(","))

    //像这种RDD中的元素是(key,value)类型的，我们将这种RDD称之为键值对RDD(kv格式RDD)
    val clazzWithAgeRDD: RDD[(String, Int)] = arrayRDD.map {
      case Array(_, _, age: String, _, clazz: String) =>
        (clazz, age.toInt)
    }

    /**
     * groupBy算子的使用
     *
     * 1、groupBy的算子，后面的分组条件是我们自己指定的
     * 2、spark中groupBy之后的，所有值会被封装到一个Iterable迭代器中存储（与scala中不同）
     * 输出：
     * (理科二班,22.556962025316455)
     * (文科三班,22.680851063829788)
     * (理科四班,22.63736263736264)
     * (理科一班,22.333333333333332)
     * (文科五班,22.30952380952381)
     */
    // val map: Map[String, List[Score]] = scoreList.groupBy((s: Score) => s.id)
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = clazzWithAgeRDD.groupBy(_._1)
//    groupRDD.foreach(println)

    val resKvRDD: RDD[(String, Double)] = groupRDD.map((kv: (String, Iterable[(String, Int)])) => {
      val clazz: String = kv._1
      val avgAge: Double = kv._2.map(_._2).sum.toDouble / kv._2.size

      (clazz, avgAge)
    })
    resKvRDD.foreach(println)

//    while (true){
//
//    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

6)groupByKey इति

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo6GroupByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("groupByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    val linesRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    //求每个班级的平均年龄
    val arrayRDD: RDD[Array[String]] = linesRDD.map((line: String) => line.split(","))


    //像这种RDD中的元素是(key,value)类型的，我们将这种RDD称之为键值对RDD(kv格式RDD)
    val clazzWithAgeRDD: RDD[(String, Int)] = arrayRDD.map {
      case Array(_, _, age: String, _, clazz: String) =>
        (clazz, age.toInt)
    }

    /**
     * GroupByKey属于kv格式的算子，只能作用在kv格式的RDD上
     * 也就说，只有kv格式的RDD才能调用kv格式的算子
     * 输出：
     * (理科二班,22.556962025316455)
     * (文科三班,22.680851063829788)
     * (理科四班,22.63736263736264)
     * (理科一班,22.333333333333332)
     * (文科五班,22.30952380952381)
     */
    val groupByKeyRDD: RDD[(String, Iterable[Int])] = clazzWithAgeRDD.groupByKey()

    val resKvRDD2: RDD[(String, Double)] = groupByKeyRDD.map((kv: (String, Iterable[Int])) => (kv._1, kv._2.sum.toDouble / kv._2.size))
    resKvRDD2.foreach(println)

    /**
     * 面试题：spark core中 groupBy算子与groupByKey算子的区别？
     * 1、代码格式上：
     * groupBy的分组条件可以自己指定，并且绝大部分的RDD都可以调用该算子，返回的是键和元素本身组成的迭代器构成的kv格式RDD
     * groupByKey算子，只能由kv格式的RDD进行调用，分组的条件会自动根据键进行分组，不需要在自己指定，返回的是键和值组成的迭代器构成的kv格式RDD
     *
     * 2、执行shuffle数据量来看
     *  groupBy产生的shuffle数据量在一定程度上要大于groupByKey产生的shuffle数据量
     *  所以groupByKey算子的执行效率要比groupBy算子的执行效率要高
     */

    while (true) {

    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

७)reduceByKey इति

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo7ReduceByKey {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("reduceByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    val linesRDD: RDD[String] = sc.textFile("spark/data/score.txt")
    //求每个班级的平均年龄
    val arrayRDD: RDD[Array[String]] = linesRDD.map((line: String) => line.split(","))
    //分别使用groupByKey和reduceBykey计算每个学生的总分
    val idWithScoreRDD: RDD[(String, Int)] = arrayRDD.map {
      case Array(id: String, _, score: String) =>
        (id, score.toInt)
    }

    /**
     * groupByKey实现
     */
//        val kvRDD1: RDD[(String, Iterable[Int])] = idWithScoreRDD.groupByKey()
//        val resRDD1: RDD[(String, Int)] = kvRDD1.map((kv: (String, Iterable[Int])) => (kv._1, kv._2.sum))
//        resRDD1.foreach(println)

    /**
     * reduceByKey实现
     * 输出：
     * (1500100113,519)
     * (1500100724,440)
     * (1500100369,376)
     * (1500100378,402)
     * (1500100306,505)
     * (1500100578,397)
     */
    val resRDD2: RDD[(String, Int)] = idWithScoreRDD.reduceByKey((v1: Int, v2: Int) => v1 + v2)
    resRDD2.foreach(println)


    /**
     * 面试题：
     * groupByKey与reduceBykey的区别？
     * 相同点：
     * 它们都是kv格式的算子，只有kv格式的RDD才能调用
     * 不同点：
     * 1）groupByKey只是单纯地根据键进行分组，分组后的逻辑可以在后续的处理中调用其他的算子实现
     * 2）reduceByKey 相当于MR中的预聚合，所以shuffle产生的数据量要比groupByKey中shuffle产生的数据量少，效率高，速度要快一些
     * 3）groupByKey的灵活度要比reduceByKey灵活度要高，reduceBykey无法做一些复杂的操作，比如方差。但是groupByKey可以在分组之后的RDD进行方差操作
     */

    while (true){

    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

८)संयोगः

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo8Union {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("reduceByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    //parallelize：将scala的集合变成spark中的RDD
    val rdd1: RDD[(String, String)] = sc.parallelize(List(
      ("1001", "张三"),
      ("1002", "张三2"),
      ("1003", "张三3"),
      ("1004", "张三4"),
      ("1005", "张三5")
    ))
    println(s"rdd1的分区数:${rdd1.getNumPartitions}")

    val rdd2: RDD[(String, String)] = sc.parallelize(List(
      ("1006", "李四6"),
      ("1007", "李四7"),
      ("1003", "张三3"),
      ("1008", "李四8"),
      ("1009", "李四9")
    ))
    println(s"rdd2的分区数:${rdd2.getNumPartitions}")

    val rdd3: RDD[(String, Int)] = sc.parallelize(List(
      ("1006", 111),
      ("1007", 22),
      ("1003", 33),
      ("1008", 444),
      ("1009", 55)
    ))

    //两个RDD要想进行union合并，必须保证元素的格式和数据类型是一致的
    //分区数也会进行合并，最终的分区数由两个RDD总共的分区数决定
    //    rdd1.union(rdd3)
    val resRDD1: RDD[(String, String)] = rdd1.union(rdd2)
    resRDD1.foreach(println)
    println(s"resRDD1的分区数:${resRDD1.getNumPartitions}")

    /**
     * 输出：
     * rdd1的分区数:1
     * rdd2的分区数:1
     * (1001,张三)
     * (1002,张三2)
     * (1003,张三3)
     * (1004,张三4)
     * (1005,张三5)
     * (1006,李四6)
     * (1007,李四7)
     * (1003,张三3)
     * (1008,李四8)
     * (1009,李四9)
     * resRDD1的分区数:2
     */
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

९)संयोगः

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * join算子也要作用在kv格式的RDD上
 */
object Demo9Join {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("reduceByKey")

    val sc: SparkContext = new SparkContext(conf)

    //===================================================
    //parallelize：将scala的集合变成spark中的RDD
    val rdd1: RDD[(String, String)] = sc.parallelize(List(
      ("1001", "张三"),
      ("1002", "李四"),
      ("1003", "王五"),
      ("1004", "小明"),
      ("1005", "小红")
    ))

    val rdd2: RDD[(String, String)] = sc.parallelize(List(
      ("1001", "看美女"),
      ("1002", "看综艺"),
      ("1003", "看八卦"),
      ("1004", "打游戏"),
      ("1009", "学习")
    ))

    /**
     * join 内连接
     * right join 右连接
     * left join 左连接
     * full join 全连接
     */
    // join 内连接 两个rdd共同拥有的键才会进行关联
    /**
     * (1001,张三,看美女)
     * (1002,李四,看综艺)
     * (1004,小明,打游戏)
     * (1003,王五,看八卦)
     */
        val resRDD1: RDD[(String, (String, String))] = rdd1.join(rdd2)
        val resRDD2: RDD[(String, String, String)] = resRDD1.map {
          case (id: String, (name: String, like: String)) =>
            (id, name, like)
        }
        resRDD2.foreach(println)

    //right join 右连接 保证右边rdd键的完整性
    /**
     * (1001,张三,看美女)
     * (1002,李四,看综艺)
     * (1004,小明,打游戏)
     * (1009,查无此人,学习)
     * (1003,王五,看八卦)
     */
        val resRDD2: RDD[(String, (Option[String], String))] = rdd1.rightOuterJoin(rdd2)
        val resRDD3: RDD[(String, String, String)] = resRDD2.map {
          case (id: String, (Some(name), like: String)) =>
            (id, name, like)
          case (id: String, (None, like: String)) =>
            (id, "查无此人", like)
        }
        resRDD3.foreach(println)

    //left join: 左连接
    /**
     * (1005,小红,此人无爱好)
     * (1001,张三,看美女)
     * (1002,李四,看综艺)
     * (1004,小明,打游戏)
     * (1003,王五,看八卦)
     */
    val resRDD1: RDD[(String, (String, Option[String]))] = rdd1.leftOuterJoin(rdd2)
    val resRDD2: RDD[(String, String, String)] = resRDD1.map {
      case (id: String, (name: String, Some(like: String))) =>
        (id, name, like)
      case (id: String, (name: String, None)) =>
        (id, name, "此人无爱好")
    }
    resRDD2.foreach(println)


    //全连接，保证所有的键、值的完整
    /**
     * (1005,小红,此人无爱好)
     * (1001,张三,看美女)
     * (1002,李四,看综艺)
     * (1004,小明,打游戏)
     * (1009,查无此人,学习)
     * (1003,王五,看八卦)
     */
    val resRDD2: RDD[(String, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)
    val resRDD3: RDD[(String, String, String)] = resRDD2.map {
      case (id: String, (Some(name), Some(like))) =>
        (id, name, like)
      case (id: String, (Some(name), None)) =>
        (id, name, "此人无爱好")
      case (id: String, (None, Some(like))) =>
        (id, "查无此人", like)
    }
    resRDD3.foreach(println)


  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110