apache支持java吗 JavaSpark-RDD编程-常见操作、持久化、函数传递

什么叫做RDD（Resilient Distributed Dataset）？

RDD有什么样的特性？

RDD是Spark的核心抽象，全程弹性分布式数据集

RDD在抽象上来说是一种数据集合，包含数据。它是被分区的。分为多个分区，每个分区分布在集群的不同节点，从而RDD中的数据可以被并行操作

RDD通常由文件（HDFS或Hive表来创建），应用程序中的集合

RDD的数据通常是存放在内存中的，内存资源不足时，spark会自动将数据写入磁盘，自动进行内存和磁盘之间的权衡和切换机制

RDD基础

spark中的RDD是一个不可变的分布式对象集合，可以包含Java、python、scala中的任意类型的对象，甚至可以包含用户自定义的对象

创建RDD：读取外部数据，驱动器程序里分发驱动器程序中的对象集合（list和set）

//外部数据读取
avaRDD<String> testFile = js.textFile("G:/sparkRS/readtest.txt");
//集合读取
JavaRDD<String> word = jsc.parallelize(data,5);

这些创建的RDD支持两种操作：转化操作（由一个RDD生成行的RDD）和行动操作（对一个RDD计算出结果）

他们区别在于spark计算方式不同，转化是惰性计算，这在大数据领域很有道理

如果需要重用一个RDD，可使用RDD.persist方法让spark把它缓存下来

数据可以持久化到许多地方

在进行一次RDD持久化操作过后，spark把RDD的内容存在内存中，之后的行动操作可重用

默认不进行持久化（对大数据集很有作用）

如果补充用该RDD，就没有必要浪费空间直接遍历结果

在任何时候都能进行重算是我们把RDD描述为‘弹性’的原因

节点故障重算也是利用这个特性

cache与使用persist是一样的

创建RDD

把程序中一个已有的集合传给 SparkContext 的 parallelize() 方法（主要用于测试）

JavaRDD<String> lines = sc.parallelize(Arrays.asList("pandas", "i like pandas"));

更常用的方式是从外部读取数据来创建RDD

avaRDD<String> testFile = js.textFile("G:/sparkRS/readtest.txt");

RDD操作

转化操作：返回一 个新的 RDD 的操作，比如 map() 和 filter()，

行动操作：向驱动器程序返回结果或把结果写入外部系统的操作，会触发实际的计算，比如 count() 和 first()。

转化操作

只有在行动操作中用到这些 RDD 时才会被计算。许多转化操作都是针对各个元素的，这些转化操作每次只会操作 RDD 中的一个元素。不过并不是所有的转化操作都是这样的

        //filter
        SparkConf  conf = new SparkConf().setMaster("local").setAppName("My App");
        JavaSparkContext js = new JavaSparkContext(conf);

        JavaRDD<String> lines = js.parallelize(Arrays.asList("coffe","coffe","panda","monkey","tea"));
        long result = lines.filter(x-> x.contains("coffe")).count();
        System.out.println(result); 
        js.close();

        //union将两个RDD合并
                SparkConf  conf = new SparkConf().setMaster("local").setAppName("My App");
        JavaSparkContext js = new JavaSparkContext(conf);

        JavaRDD<String> lines = js.parallelize(Arrays.asList("coffe","coffe","panda","monkey","tea"));
        JavaRDD<String> lines1 = js.parallelize(Arrays.asList("coffe","coffe","panda","monkey","tea"));
        JavaRDD<String> result = lines.filter(x-> x.contains("coffe"));
        JavaRDD<String> result1 = lines1.filter(x-> x.contains("tea"));
        JavaRDD<String> outcome = result.union(result1);
        System.out.println(outcome.collect());  
        js.close();

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/sanxing/article-57960-1.html