1个
2
3
4
5
6
7
8
9
10
11

val rdd = sc.makeRDD（Array（
  （5，“汤姆”），（10，“杰德”），（3，“托尼”），（2，“杰克”）
））
rdd.sortByKey（）。foreach（println）
结果：
（2，杰克）
（3，贝）
（5，汤姆）
（10，杰德）
说明：
sortByKey（fasle）：倒序

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

对象 SortByOperator  {
  def  main（args： Array [ String ]）： Unit = {
   val conf = new  SparkConf（）.setAppName（“ TestSortBy”）.setMaster（“ local”）
    val sc = 新的 SparkContext（conf）
    val arr = 数组（
        Tuple3（190，100，“杰德”），
        Tuple3（100，202，“汤姆”），
        Tuple3（90，111，“托尼”）
    ）
    val rdd = sc.parallelize（arr）
    rdd.sortBy（_._ 1）.foreach（println）
    / *（90,111，托尼）
       （100,202，汤姆）
       （190,100，Jed）
     * /
    rdd.sortBy（_._ 2）.foreach（println）
    / *（190,100，Jed）
       （90,111，托尼）
       （100,202，汤姆）
     * /
    rdd.sortBy（_._ 3）.foreach（println）
    / *
       （190,100，Jed）
       （100,202，汤姆）
       （90,111，托尼）
     * /
    sc.stop（）;
  }
}

1个
2
3
4
5
6
7
8
9
10
11
12
13

val rdd = sc.makeRDD（Array（
      “你好”，
      “你好”，
      “你好”，
      “世界”
））
valdistingRDD = rdd
      .map {（_，1）}
      .reduceByKey（_ + _）
      .map（_._ 1）
distantRDD.foreach {println}
等价于：
rdd.distinct（）。foreach {println}

1个
2
3

选择 * 从 TableA
INNER  JOIN表B
开启 TableA.name = TableB.name

1个
2
3

选择 * 从 TableA
LEFT  OUTER  JOIN表B
开启 TableA.name = TableB.name

1个
2
3

选择 * 从 TableA
RIGHT  OUTER  JOIN表B
开启 TableA.name = TableB.name

1个
2
3

选择 * 从 TableA
FULL  OUTER  JOIN表B
开启 TableA.name = TableB.name

1个
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18岁
19
20
21
22

val nameList = 列表（
      （1，“ Jed”），
      （2，“汤姆”），
      （3，“鲍勃”），
      （4，“托尼”）
）
   
val SalonArr = 数组（
      （1，8000），
      （2，6000），
      （3，5000）
）
/ *
 * parallelize（Seq [T]，Int num）
 *使用指定的集合（可以是List，Array等）来创建RDD
 * num指定RDD的分区数，交替为1
 *这个方法经常用于测试环境
 * join产生的RDD的分区数由分区数最多的父RDD决定
 * /
val nameRDD = sc.parallelize（nameList，2）
val salalRDD = sc.parallelize（salaryArr，3）

1个
2
3
4
5
6
7

val joinRDD =名称RDD.join（salaryRDD）
joinRDD.foreach（x => {
    值 ID = x._1
    值名称= x._2._1
    瓦尔工资= x._2._2
    println（id + “ \ t” +名称+ “ \ t” +薪水）
}）

1个
2
3
4
5
6
7

val leftOuterJoinRDD = nameRDD.leftOuterJoin（salaryRDD）
leftOuterJoinRDD.foreach（x => {
      值 ID = x._1
      值名称= x._2._1
      瓦尔工资= x._2._2
      println（id + “ \ t” +名称+ “ \ t” +薪水）
}）

1个
2
3
4
5
6
7

val rightOuterJoinRDD = nameRDD.rightOuterJoin（salaryRDD）
rightOuterJoinRDD.foreach（x => {
      值 ID = x._1
      值名称= x._2._1
      瓦尔工资= x._2._2
      println（id + “ \ t” +名称+ “ \ t” +薪水）
}）

1个
2
3
4
5
6
7

val fullOuterJoinRDD = nameRDD.fullOuterJoin（salaryRDD）
fullOuterJoinRDD.foreach（x => {
      值 ID = x._1
      值名称= x._2._1
      瓦尔工资= x._2._2
      println（id + “ \ t” +名称+ “ \ t” +薪水）
}）

1个
2
3

val rdd1 = sc.makeRDD（1至10）
val rdd2 = sc.parallelize（11到20）
rdd1.union（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

val arr = 数组（“汤姆”，“鲍勃”，“托尼”，“杰里”）
//把4条数据分到两个分区中
val rdd = sc.parallelize（arr，2）
   
/ *
 *模拟把RDD中的元素写入数据库的过程
 * /
rdd.map（x => {
  println（“创建数据库连接...”）
  println（“写入数据库...”）
  println（“关闭数据库连接...”）
  println（）
}）。计数（）
结果：
创建数据库连接...
写入数据库...
关闭数据库连接...
创建数据库连接...
写入数据库...
关闭数据库连接...
创建数据库连接...
写入数据库...
关闭数据库连接...
创建数据库连接...
写入数据库...
关闭数据库连接...

1个
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18岁
19
20
21

/ *
 *将RDD中的数据写入到数据库中，绝大部分使用mapPartitions算子来实现
 * /
rdd.mapPartitions（x => {
  println（“创建数据库”）
  val list = new  ListBuffer [ String ]（）
  而（x.hasNext）{
    //编写数据库
    列表+ = x.next（）+ “：写入数据库”
  }
  //执行SQL语句批量插入
  list.iterator
}）foreach（println）
结果：
创建数据库
汤姆：编写数据库
鲍勃：编写数据库
创建数据库
托尼：编写数据库
杰瑞：编写数据库