spark: core相关

Author: Yang-xingchen

middleware/spark/SparkCore.md

@@ -0,0 +1,55 @@
+# Job/Stage/Task
+- Job: 作业，完整的执行流程。即count(Job)=count(行动算子)。
+- Stage: 阶段，每次读写算做一个阶段。即count(Stage)=count(Shuffle)+1。各阶段串行执行。
+- Task: 任务，每个阶段最后RDD分区数量总和。
+
+Job 1=(1..n) Stage 1=(1..n) Task
+
+# Partition
+分区。将数据分成多个区，各区数据不重复。
+k-v类型相同key放在同一个区。
+Shuffle操作会重新分区。
+
+# Shuffle
+将分区内数据重新打乱分发操作称为Shuffle。Shuffle操作是主要的资源消耗原因。
+- 可改变分区数目。
+- Shuffle一定会落盘。
+- 需等待所有数据Shuffle操作执行完毕后才继续执行后续操作。
+
+# RDD
+Resilient Distributed Datasets弹性分布式数据集。
+
+- 封装单个简单的数据处理，复杂处理需多个RDD组合
+- 不保存具体数据值
+- 可处理多个分区
+
+## 数据处理分类
+- value: 每条数据只有值，处理也是按值处理。
+- key-value: 每条数据包含键值对，可按键值处理。
+
+## 算子(方法处理分类)
+指RDD内的方法。对数据进行操作。
+- 转换算子: 对数据进行转化操作，返回结果还是RDD。 一般不会创建执行任务。
+- 行动算子: 对数据进行收集操作，返回结果不是RDD。 会创建并执行任务。
+
+# 依赖
+相邻RDD间数据分区的关系。下游RDD依赖直接上游RDD。
+
+## 窄依赖
+- 上游RDD数据被一个RDD独享。
+- 上游 (1..n)=1 下游。
+- 不会执行shuffle操作。
+
+## 宽依赖
+- 上游RDD数据被多个RDD共享。
+- 上游 (1..n)=n 下游。
+- 会执行shuffle操作。
+
+# 持久化
+避免重复计算，将计算中间结果保存。
+- cache: 缓存。不切断血缘关系，存储在内存。
+- checkPoint: 检测点。切断血缘关系，通常存储在HDFS。会重新计算。
+
+# Broadcast
+广播变量，当task使用到共享变量时，避免多次传输时使用。
+该方法可将共享变量按工作节点分发。(默认采用task分发)

middleware/spark/src/main/java/base/CacheRdd.java

@@ -0,0 +1,56 @@
+package base;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.Collections;
+
+public class CacheRdd {
+
+    /**
+     * map + plus  + zip
+     *     | multi |
+     */
+    public static void main(String[] args) {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local[2]")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+
+//        JavaRDD<Integer> map = javaSparkContext.parallelize(Collections.singletonList(1))
+//                .map(i -> {
+//                    // 执行2次
+//                    System.out.println("1. [" + Thread.currentThread() + "]: " + i);
+//                    return i;
+//                });
+        JavaRDD<Integer> map = javaSparkContext.parallelize(Collections.singletonList(1))
+                .map(i -> {
+                    // 执行1次
+                    System.out.println("1. [" + Thread.currentThread() + "]: " + i);
+                    return i;
+                })
+                .cache();
+
+        JavaRDD<Integer> plus = map.map(i -> {
+            // 执行1次
+            System.out.println("2.1 [" + Thread.currentThread() + "]: " + i);
+            return i + 1;
+        });
+        JavaRDD<Integer> multi = map.map(i -> {
+            // 执行1次
+            System.out.println("2.2 [" + Thread.currentThread() + "]: " + i);
+            return i * 2;
+        });
+        JavaRDD<Integer> zip = plus.zip(multi).map(tuple -> {
+            // 执行1次
+            System.out.println("3. [" + Thread.currentThread() + "]: " + (tuple._1() + tuple._2()));
+            return tuple._1() + tuple._2();
+        });
+        zip.collect().forEach(System.out::println);
+        spark.close();
+    }
+
+}

middleware/spark/src/main/java/base/CheckPointRdd.java

@@ -0,0 +1,54 @@
+package base;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.Optional;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class CheckPointRdd {
+
+    public static final List<Integer> DATA = Stream
+            .of(1, 2, 3, 4, 5, 6, 7, 8, 9)
+            .collect(Collectors.toList());
+
+    public static void main(String[] args) throws Exception {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+//        // 需提前创建目录
+        javaSparkContext.setCheckpointDir("checkpoint");
+        JavaRDD<Integer> cache = javaSparkContext
+                .parallelize(DATA)
+                .map(i -> {
+                    System.out.println("map: " + i);
+                    return i;
+                })
+                .cache();
+        cache.checkpoint();
+        cache.collect().forEach(System.out::println);
+
+        // 获取checkpoint输出文件
+        Optional<String> checkpointFile = cache.getCheckpointFile();
+        System.out.println("file:" + checkpointFile.orElse(null));
+
+        javaSparkContext = new JavaSparkContext(spark.sparkContext());
+        // 读取checkpoint
+        javaSparkContext.checkpointFile(checkpointFile.get())
+                .collect()
+                .forEach(System.out::println);
+
+        // job: 3
+        System.out.println("open: http://localhost:4040");
+        TimeUnit.HOURS.sleep(1);
+        spark.close();
+    }
+
+}

middleware/spark/src/main/java/base/Partition.java

@@ -0,0 +1,43 @@
+package base;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+import scala.Tuple2;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class Partition {
+
+    public static final List<Integer> DATA = Stream
+            .of(1, 2, 3, 4, 5, 6, 7, 8, 9)
+            .collect(Collectors.toList());
+
+    public static void main(String[] args) {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+        javaSparkContext
+                .parallelize(DATA)
+                .groupBy(i -> i)
+                .groupByKey(new org.apache.spark.Partitioner() {
+                    @Override
+                    public int numPartitions() {
+                        return 3;
+                    }
+
+                    @Override
+                    public int getPartition(Object key) {
+                        return key.hashCode() % 3;
+                    }
+                })
+                .map(Tuple2::_1)
+                .saveAsTextFile("output");
+        spark.close();
+    }
+
+}

middleware/spark/src/main/java/base/StagePartition.java

@@ -0,0 +1,59 @@
+package base;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class StagePartition {
+
+    public static final List<Integer> DATA = Stream
+            .of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12)
+            .collect(Collectors.toList());
+
+    public static void main(String[] args) throws Exception {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local[12]")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+        javaSparkContext
+                // stage1: 2
+                .parallelize(DATA, 2)
+                .map(i -> {
+                    // 2 thread
+                    System.out.println("parallelize[" + Thread.currentThread().hashCode() + "]: " + i);
+                    return i;
+                })
+                // stage2: 6
+                .repartition(12)
+                .map(i1 -> {
+                    // 6 thread
+                    System.out.println("repartition[" + Thread.currentThread().hashCode() + "]: " + i1);
+                    return i1;
+                })
+                .coalesce(6)
+                .map(i11 -> {
+                    // 6 thread
+                    System.out.println("coalesce1[" + Thread.currentThread().hashCode() + "]: " + i11);
+                    return i11;
+                })
+                // stage3: 3
+                .coalesce(3, true)
+                .map(i11 -> {
+                    // 3 thread
+                    System.out.println("coalesce2[" + Thread.currentThread().hashCode() + "]: " + i11);
+                    return i11;
+                })
+                .collect()
+                .forEach(System.out::println);
+        System.out.println("open: http://localhost:4040");
+        TimeUnit.HOURS.sleep(1);
+        spark.close();
+    }
+
+}