spark: 基础RDD

Author: Yang-xingchen

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

@@ -0,0 +1,47 @@
+package base;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class BaseRdd {
+
+    public static final List<String> DATA = Stream
+            .of("Abc", "Abcdef", "bC", "Dd","eC", "dD")
+            .collect(Collectors.toList());
+
+    public static void main(String[] args) {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+        List<String> res = javaSparkContext
+                .parallelize(DATA)
+                // 过滤: 移除 Abcdef
+                .filter(s -> s.length() < 5)
+                // 映射: 转大写
+                .map(String::toUpperCase)
+                // 去重: 合并 Dd dD
+                .distinct()
+                // 展开: 按字符串长度复制多份
+                .flatMap(s -> {
+                    List<String> list = new ArrayList<>(s.length());
+                    for (int i = 0; i < s.length(); i++) {
+                        list.add(s);
+                    }
+                    return list.iterator();
+                })
+                // 排序: 字典序
+                .sortBy(s -> s, true, 2)
+                .collect();
+        res.forEach(System.out::println);
+        spark.close();
+    }
+
+}

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

@@ -0,0 +1,47 @@
+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 PairRdd {
+
+    public static final List<Tuple2<String, Integer>> DATA = Stream
+            .of(
+                    new Tuple2<>("ABC", 1),
+                    new Tuple2<>("abc", 2),
+                    new Tuple2<>("Abc", 3),
+                    new Tuple2<>("Abc", 3),
+                    new Tuple2<>("Abc", 4),
+                    new Tuple2<>("ABc", 3),
+                    new Tuple2<>("aBc", 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());
+        List<Tuple2<String, Integer>> res = javaSparkContext
+                .parallelizePairs(DATA)
+                // 过滤: 移除 ("aBc", 9)
+                .filter(t -> t._2() < 5)
+                // 去重: K-V相同才去重
+                .distinct()
+                // 映射值: 双倍
+                .mapValues(i -> i * 2)
+                // 排序: 按key排
+                .sortByKey()
+                .collect();
+        res.forEach(System.out::println);
+        spark.close();
+    }
+
+}

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

@@ -0,0 +1,39 @@
+package base;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class WordCount {
+
+    public static final List<String> DATA = Stream
+            .of("Abc", "Abcdef", "bC", "Dd","eC", "dD", "Abcdef", "bC", "bC", "Dd","eC", "dD", "Abcdef", "bC")
+            .collect(Collectors.toList());
+
+    public static void main(String[] args) {
+        SparkSession spark = SparkSession
+                .builder()
+                .appName("test")
+                .master("local")
+                .getOrCreate();
+        JavaSparkContext javaSparkContext = new JavaSparkContext(spark.sparkContext());
+        Map<String, Integer> res = javaSparkContext
+                .parallelize(DATA)
+                .groupBy(s -> s)
+                .mapValues(strings -> {
+                    int count = 0;
+                    for (String s : strings) {
+                        count++;
+                    }
+                    return count;
+                })
+                .collectAsMap();
+        res.forEach((k, v) -> System.out.println(k + "\t" + v));
+        spark.close();
+    }
+
+}