Spark RDD使用教程_头歌实验 spark rdd操作数据库

1、创建RDD的三种方式

Spark提供三种创建RDD方式：集合、本地文件、HDFS文件。详细可以查看RDD和pair RDD文档
使用例子SparkRdd.java

ppackage com.penngo.rdd;

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.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 *
 */
public class SparkRdd {
    public static void main(String[] args) {
        //windows下调试spark需要使用https://github.com/steveloughran/winutils
        System.setProperty("hadoop.home.dir", "D:\\hadoop\\hadoop-3.3.1");
        System.setProperty("HADOOP_USER_NAME", "root");

        SparkSession spark = SparkSession
                .builder()
                .appName("List2Rdd")
                .master("local[*]")
                .getOrCreate();
        JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());

        // 集合转成RDD
        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5,1,2);
        JavaRDD<Integer> rdd1 = sc.parallelize(data);
        int sum1 = rdd1.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        System.out.println("sum1============"+sum1);

        // 读取本地文件/hadoop文件转成RDD
        String path = "D:\\project\\data.txt";
        //path = "hdfs://testspark:9000/data.txt"
        JavaRDD<String> rddLine = sc.textFile(path, 2);

        JavaRDD<Integer> rdd2 =  rddLine.flatMap((FlatMapFunction<String, Integer>) s -> {
            String[] strs = s.split(",");
            List<Integer> list = new ArrayList<>();
            for(String str:strs){
                list.add(Integer.valueOf(str.trim()));
            }
            return list.iterator();
        });

        int sum2 = rdd2.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        System.out.println("sum2============"+sum2);

        JavaPairRDD<String, Integer> pairs = rdd2.mapToPair(s -> new Tuple2("num_" + s, 1));
        JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);
        counts.foreach(tp2->{
            System.out.println(tp2._1 + "=" + tp2._2);
        });
        spark.stop();
    }
}

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

2、Spark对RDD的操作

Spark对RDD的操作可以整体分为两类：
Transformation（转换）：表示是针对RDD中数据的转换操作，主要会针对已有的RDD创建一个新的RDD：常见的有map、flatMap、filter等等。
Action（执行）表示是触发任务执行的操作，主要对RDD进行最后的操作，比如遍历、reduce、保存到文件等，并且还可以把结果返回给Driver程序

2.1、Transformations（转换）

2.2、Actions（动作）

参考自官方文档