线下课：RDD的创建、RDD的操作、RDD重分区

src/main/java/org/example/core/GroupByKeyDemo.java

@@ -0,0 +1,55 @@
+package org.example.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.Function;
+import scala.Tuple2;
+
+import java.util.Arrays;
+
+/**
+ * Hello world!
+ */
+public class GroupByKeyDemo {
+    public static void main(String[] args) throws InterruptedException {
+        SparkConf sparkConf = new SparkConf()
+                .setMaster("local")
+                .setAppName("my spark app");
+        JavaSparkContext sc = new JavaSparkContext(sparkConf);
+        // 先当成集合来看
+        JavaRDD<String> myTextFile = sc.textFile("src/main/resources/word.txt");
+
+        // (hello,[1,1,1,1,1])
+        JavaPairRDD<String, Iterable<Integer>> groupByRet = myTextFile.flatMap(line -> Arrays.asList(line.split(" ")).iterator())
+                .mapToPair(word -> new Tuple2<String, Integer>(word, 1))
+                .groupByKey();
+        JavaPairRDD<String, Integer> result = groupByRet.mapValues(new Function<Iterable<Integer>, Integer>() {
+            @Override
+            public Integer call(Iterable<Integer> v1) throws Exception {
+                int sum = 0;
+                for (Integer i : v1) {
+                    sum = sum + i;
+                }
+                return sum;
+            }
+        });
+
+        Thread.sleep(10000000L);
+
+
+
+
+    }
+}
+
+
+
+
+
+
+
+
+
+

src/main/java/org/example/core/MyPartitioner.java

@@ -0,0 +1,23 @@
+package org.example.core;
+
+import org.apache.spark.Partitioner;
+
+public class MyPartitioner extends Partitioner {
+//    private int partitionNum = 0;
+
+    // 我想分成奇数和偶数，所以分区数是2
+    @Override
+    public int numPartitions() {
+        return 2;
+    }
+
+    @Override
+    public int getPartition(Object key) {
+        if (key instanceof Integer) {
+            Integer number = (Integer) key;
+            return number % 2;
+        } else {
+            throw new IllegalArgumentException("key must be an integer");
+        }
+    }
+}

src/main/java/org/example/core/RddCreateDemo.java

@@ -0,0 +1,40 @@
+package org.example.core;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class RddCreateDemo {
+    public static void main(String[] args) throws InterruptedException {
+
+        SparkConf sparkConf = new SparkConf()
+                .setMaster("local[1]") // local模式采用 多线程 模拟 多进程  （executor 个数 、partition）
+                .setAppName("my spark app");
+        JavaSparkContext sc = new JavaSparkContext(sparkConf);
+
+        // 1. 从集合中创建
+        List<String> data = new ArrayList<>();
+        data.add("a");
+        data.add("b");
+        data.add("c");
+        data.add("d");
+
+        JavaRDD<String> rdd = sc.parallelize(data);
+        // 调整并行度
+        JavaRDD<String> rdd2 = rdd.repartition(2);
+// 在action动作之前，是不是也print不出新的rdd对象
+        JavaRDD<String> map = rdd2.map(word -> "test:" + word);
+        rdd2.saveAsTextFile("src/main/resources/rddtest2"+System.currentTimeMillis());
+        map.collect();
+//        // 2. 从外部获取（可以是文件，也可以是jdbc【sparkSession】）
+//        JavaRDD<String> myTextFile = sc.textFile("src/main/resources/word.txt");
+//        JavaRDD<String> myHdfsFile = sc.textFile("hdfs://192.168.56.101:8020/word.txt");
+//        myTextFile.saveAsTextFile("src/main/resources/word");
+
+
+        Thread.sleep(10000000L);
+    }
+}

src/main/java/org/example/core/RepartitionVsCoaleasce.java

@@ -0,0 +1,77 @@
+package org.example.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 scala.Tuple2;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+/*
+repartition和coalesce都可以用来减少分区，区别在于要不要打散数据。另外repartition可以用来放大分区
+repartition的底层是通过 带 shuffle 的coalesce 来实现的。
+  def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
+    coalesce(numPartitions, shuffle = true)
+  }
+shuffle的话
+    缺点：带来一定的网络开销，可能会比较低效
+    优点：可以平均的打散数据，避免出现数据倾斜
+
+coalesce通过合并相邻的物理分区来减少分区数量
+    优点：减少不必要的网络开销，高效
+    缺点：不可以平均的打散数据，可能出现数据倾斜
+
+ */
+public class RepartitionVsCoaleasce {
+    public static void main(String[] args) throws InterruptedException {
+
+
+        SparkConf sparkConf = new SparkConf()
+                .setMaster("local[2]")
+                .setAppName("my spark app");
+        JavaSparkContext sc = new JavaSparkContext(sparkConf);
+
+        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+        JavaRDD<Integer> rdd = sc.parallelize(data, 4);
+        rdd.cache();
+
+        System.out.println("初始化分区的数据分布情况：");
+        printRdd(rdd);
+
+        // 1. repartition
+        JavaRDD<Integer> repartitionRdd = rdd.repartition(2);
+        System.out.println("repartition分区后的数据分布情况：");
+        printRdd(repartitionRdd);
+
+        // 2. coalesce
+        JavaRDD<Integer> coalesceRdd = rdd.coalesce(2,false);
+        System.out.println("coalesce分区后的数据分布情况：");
+        printRdd(coalesceRdd);
+
+        // 3. 自定义分区器
+        JavaPairRDD<Integer, Object> pairRDD = rdd.mapToPair(num -> new Tuple2<>(num, null));
+        JavaRDD<Integer> customerPartitionRdd = pairRDD.partitionBy(new MyPartitioner()).map(tuple2 -> tuple2._1);
+        System.out.println("customerPartitionRdd分区后的数据分布情况：");
+        printRdd(customerPartitionRdd);
+
+        Thread.sleep(10000000L);
+
+        sc.stop();
+
+
+    }
+
+    private static void printRdd(JavaRDD<Integer> rdd) {
+        Iterator<List<List<Integer>>> partitions = rdd.glom().collect().stream().map(Arrays::asList).iterator();
+        int i = 0;
+        while (partitions.hasNext()) {
+            i++;
+            List<List<Integer>> partiton = partitions.next();
+            System.out.println("分区 【"+i+"】 ：" + partiton.get(0));
+        }
+    }
+}

src/main/scala/org/example/App.scala

@@ -1,4 +1,4 @@
-package com.example.spark
+package org.example
 
 import org.apache.spark.{SparkConf, SparkContext}
 
@@ -8,8 +8,8 @@ object App {
     val sparkConf = new SparkConf()
       .setMaster("local[2]")
       .setAppName("scala spark app")
-      .set("spark.driver.host", "127.0.0.1")
-      .set("spark.driver.bindAddress", "127.0.0.1")
+//      .set("spark.driver.host", "127.0.0.1")
+//      .set("spark.driver.bindAddress", "127.0.0.1")
 
     val sc = new SparkContext(sparkConf)
 

src/main/scala/org/example/core/GroupByKeyDemo.scala

@@ -0,0 +1,25 @@
+package org.example.core
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object GroupByKeyDemo {
+  def main(args: Array[String]): Unit = {
+
+    val conf = new SparkConf()
+      .setMaster("local")
+      .setAppName("my spark app")
+
+    val sc = new SparkContext(conf)
+
+    val myTextFile = sc.textFile("src/main/resources/word.txt")
+
+    val result = myTextFile
+      .flatMap(_.split(" "))
+      .map(word => (word, 1))
+      .groupByKey()
+      .mapValues(_.sum)
+
+    result.collect().foreach(println)
+    sc.stop()
+  }
+}

src/main/scala/org/example/core/MyPartitioner.scala

@@ -0,0 +1,19 @@
+package org.example.core
+
+import org.apache.spark.Partitioner
+
+class MyPartitioner extends Partitioner {
+
+  // 我想分成奇数和偶数，所以分区数是 2
+  override def numPartitions = 2
+
+  override def getPartition(key: Any): Int = {
+    key match {
+      case number: Int =>
+        number % 2
+
+      case _ =>
+        throw new IllegalArgumentException("key must be an integer")
+    }
+  }
+}

src/main/scala/org/example/core/RddCreateDemo.scala

@@ -0,0 +1,41 @@
+package org.example.core
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+import java.io.File
+
+object RddCreateDemo {
+  def main(args: Array[String]): Unit = {
+
+    val sc = new SparkContext(new SparkConf().setAppName("rdd").setMaster("local[2]"))
+
+    // 1. 从集合中创建 RDD
+    val data = List("a", "b", "c", "d")
+
+    val rdd2 = sc.parallelize(data, 4)
+
+    // 调整并行度
+    //    val rdd2 = rdd.repartition(1)
+
+    // 转换算子：map
+    val map = rdd2.map(word => "test:" + word)
+
+    // Action：保存 RDD
+    if (!new File("src/main/resources/rddtest").exists()) {
+      rdd2.saveAsTextFile("src/main/resources/rddtest")
+    }
+
+    // Action：收集结果
+    map.collect()
+
+    // 2. 从外部获取
+    val myTextFile = sc.textFile("src/main/resources/word.txt", 2)
+    // val myHdfsFile = sc.textFile("hdfs://192.168.56.101:8020/word.txt")
+    if (!new File("src/main/resources/word").exists()) {
+      myTextFile.saveAsTextFile("src/main/resources/word")
+    }
+
+    //    Thread.sleep(10000000L)
+    sc.stop()
+  }
+}

src/main/scala/org/example/core/RepartitionVsCoalesce.scala

@@ -0,0 +1,67 @@
+package org.example.core
+
+import org.apache.spark.{SparkConf, SparkContext}
+
+object RepartitionVsCoalesce {
+
+  def main(args: Array[String]): Unit = {
+
+    val conf = new SparkConf()
+      .setMaster("local[2]")
+      .setAppName("my spark app")
+
+    val sc = new SparkContext(conf)
+
+    val data = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+
+    // 创建 4 个分区
+    val rdd = sc.parallelize(data, 4)
+
+    rdd.cache()
+
+    println("初始化分区的数据分布情况：")
+    printRdd(rdd)
+
+    // 1. repartition
+    // 增加或减少分区都会 shuffle
+    val repartitionRdd = rdd.repartition(2)
+
+    println("repartition 分区后的数据分布情况：")
+    printRdd(repartitionRdd)
+
+    // 2. coalesce
+    // 减少分区，第二个参数 false 表示不进行 shuffle
+    val coalesceRdd = rdd.coalesce(2, shuffle = false)
+
+    println("coalesce 分区后的数据分布情况：")
+    printRdd(coalesceRdd)
+
+    // 3. 自定义分区器
+    // 先把普通 RDD 转成 Pair RDD
+    val pairRDD = rdd.map((_, null))
+
+    // 使用自定义分区器
+    val customerPartitionRdd =
+      pairRDD
+        .partitionBy(new MyPartitioner)
+        .map(_._1)
+
+    println("customerPartitionRdd 分区后的数据分布情况：")
+    printRdd(customerPartitionRdd)
+
+    sc.stop()
+  }
+
+
+  def printRdd(rdd: org.apache.spark.rdd.RDD[Int]): Unit = {
+
+    val partitions = rdd.glom().collect()
+
+    partitions.zipWithIndex.foreach {
+      case (partition, index) =>
+        println(
+          s"分区【${index + 1}】：${partition.mkString("[", ", ", "]")}"
+        )
+    }
+  }
+}

src/main/scala/org/example/test.scala

@@ -1,4 +1,4 @@
-package com.example.spark
+package org.example
 
 import org.apache.spark.{SparkConf, SparkContext}
 
@@ -7,8 +7,9 @@ object test {
     val sc = new SparkContext(new SparkConf()
       .setMaster("local")
       .setAppName("test")
-      .set("spark.driver.host", "127.0.0.1")
-      .set("spark.driver.bindAddress", "127.0.0.1"))
+//      .set("spark.driver.host", "127.0.0.1")
+//      .set("spark.driver.bindAddress", "127.0.0.1")
+    )
     sc.textFile("src/main/resources/word.txt").foreach(println)
   }
 }