Spark RDD常用算子操作（十一） RDD Action 保存操作saveAsTextFile,saveAsSequenceFile,saveAsObjectFile,saveAsHadoopFile 等

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

RDD Action 保存操作

关键字:Spark算子、Spark函数、Spark RDD行动Action、Spark RDD存储操作、saveAsTextFile、saveAsSequenceFile、saveAsObjectFile,saveAsHadoopFile、saveAsHadoopDataset,saveAsNewAPIHadoopFile、saveAsNewAPIHadoopDataset

saveAsTextFile

def saveAsTextFile(path: String): Unit

def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit

saveAsTextFile用于将RDD以文本文件的格式存储到文件系统中。

codec参数可以指定压缩的类名。

1
2
3
4
5
6
7
8
var rdd1 = sc.makeRDD(1 to 10,2)
scala> rdd1.saveAsTextFile("hdfs://cdh5/tmp/lxw1234.com/") //保存到HDFS
hadoop fs -ls /tmp/lxw1234.com
Found 2 items
-rw-r--r--   2 lxw1234 supergroup        0 2015-07-10 09:15 /tmp/lxw1234.com/_SUCCESS
-rw-r--r--   2 lxw1234 supergroup        21 2015-07-10 09:15 /tmp/lxw1234.com/part-00000

hadoop fs -cat /tmp/lxw1234.com/part-00000

注意：如果使用rdd1.saveAsTextFile(“file:///tmp/lxw1234.com”)将文件保存到本地文件系统，那么只会保存在Executor所在机器的本地目录。
指定压缩格式保存

1
2
3
4
5
6
7
rdd1.saveAsTextFile("hdfs://cdh5/tmp/lxw1234.com/",classOf[com.hadoop.compression.lzo.LzopCodec])

hadoop fs -ls /tmp/lxw1234.com
-rw-r--r--   2 lxw1234 supergroup    0 2015-07-10 09:20 /tmp/lxw1234.com/_SUCCESS
-rw-r--r--   2 lxw1234 supergroup    71 2015-07-10 09:20 /tmp/lxw1234.com/part-00000.lzo

hadoop fs -text /tmp/lxw1234.com/part-00000.lzo

阅读全文 »

Spark RDD常用算子操作（十） PairRDD的Action操作countByKey, collectAsMap

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

PairRDD的Action操作countByKey, collectAsMap

countByKey

def countByKey(): Map[K, Long]
以RDD{(1, 2),(2,4),(2,5), (3, 4),(3,5), (3, 6)}为例 rdd.countByKey会返回{(1,1),(2,2),(3,3)}
scala例子

1
2
3
4
scala> val rdd = sc.parallelize(Array((1, 2),(2,4),(2,5), (3, 4),(3,5), (3, 6)))

scala> val countbyKeyRDD = rdd.countByKey()
countbyKeyRDD: scala.collection.Map[Int,Long] = Map(1 -> 1, 2 -> 2, 3 -> 3)

java例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
    JavaRDD<Tuple2<Integer, Integer>> tupleRDD =
            sc.parallelize(Arrays.asList(new Tuple2<>(1, 2),
            new Tuple2<>(2, 4),
            new Tuple2<>(2, 5),
            new Tuple2<>(3, 4),
            new Tuple2<>(3, 5),
            new Tuple2<>(3, 6)));
    JavaPairRDD<Integer, Integer> mapRDD = JavaPairRDD.fromJavaRDD(tupleRDD);
    //countByKey
    Map<Integer, Object> countByKeyRDD = mapRDD.countByKey();
    for (Integer i:countByKeyRDD.keySet()) {
        System.out.println("("+i+", "+countByKeyRDD.get(i)+")");
    }
/*
输出  
(1, 1)
(3, 3)
(2, 2)

*/

阅读全文 »

Spark RDD常用算子操作（九）基本的Action操作 first, take, collect, count, countByValue, reduce, aggregate, fold,top

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

基本的Action操作 first, take, collect, count, countByValue, reduce, aggregate, fold,top

first

返回第一个元素
scala

1
2
3
4
scala> val rdd = sc.parallelize(List(1,2,3,3))

scala> rdd.first()
res1: Int = 1

java

1
2
JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3, 3));
Integer first = rdd.first();

阅读全文 »

Spark RDD常用算子操作（八） 键值对关联操作 subtractByKey, join,fullOuterJoin, rightOuterJoin, leftOuterJoin

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

键值对关联操作 subtractByKey, join,fullOuterJoin, rightOuterJoin, leftOuterJoin

github: https://github.com/zhaikaishun/spark_tutorial/tree/master/src/main/java/com/spark/rdd_tutorial/tutorial8
先从spark-learning中的一张图大致了解其功能
键值对操作

阅读全文 »

Spark RDD常用算子操作（七） 键值对分组操作 groupByKey，cogroup

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

键值对分组操作 groupByKey，cogroup

groupByKey

1
2
3
4
5
def groupByKey(): RDD[(K, Iterable[V])]

def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]

def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]

groupByKey会将RDD[key,value] 按照相同的key进行分组，形成RDD[key,Iterable[value]]的形式， 有点类似于sql中的groupby，例如类似于mysql中的group_concat
例如这个例子， 我们对学生的成绩进行分组
scala版本

1
2
3
4
5
6
7
val scoreDetail = sc.parallelize(List(("xiaoming",75),("xiaoming",90),("lihua",95),("lihua",100),("xiaofeng",85)))
scoreDetail.groupByKey().collect().foreach(println(_));
/*输出
(lihua,CompactBuffer(95, 100))
(xiaoming,CompactBuffer(75, 90))
(xiaofeng,CompactBuffer(85))
*/

java版本

1
2
3
4
5
6
7
8
9
10
JavaRDD<Tuple2<String,Float>> scoreDetails = sc.parallelize(Arrays.asList(new Tuple2("xiaoming", 75)
        , new Tuple2("xiaoming", 90)
        , new Tuple2("lihua", 95)
        , new Tuple2("lihua", 188)));
//将JavaRDD<Tuple2<String,Float>> 类型转换为 JavaPairRDD<String, Float>
JavaPairRDD<String, Float> scoreMapRDD = JavaPairRDD.fromJavaRDD(scoreDetails);
Map<String, Iterable<Float>> resultMap = scoreMapRDD.groupByKey().collectAsMap();
for (String key:resultMap.keySet()) {
    System.out.println("("+key+", "+resultMap.get(key)+")");
}

阅读全文 »

Spark RDD常用算子操作（六） 键值对聚合操作reduceByKey，foldByKey，排序操作sortByKey

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

键值对聚合操作reduceByKey，foldByKey，排序操作sortByKey

reduceByKey

1
2
3
4
5
def reduceByKey(func: (V, V) => V): RDD[(K, V)]

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]

接收一个函数，按照相同的key进行reduce操作，类似于scala的reduce的操作
例如RDD {(1, 2), (3, 4), (3, 6)}进行reduce
scala版本

1
2
3
4
5
6
    var mapRDD = sc.parallelize(List((1,2),(3,4),(3,6)))
    var reduceRDD = mapRDD.reduceByKey((x,y)=>x+y)
    reduceRDD.foreach(x=>println(x))
------输出---------
(1,2)
(3,10)

再举例
单词计数
F:\sparktest\sample.txt中的内容如下

1
2
3
4
aa bb cc aa aa aa dd dd ee ee ee ee 
ff aa bb zks
ee kks
ee  zz zks

scala版本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
    val lines = sc.textFile("F:\\sparktest\\sample.txt")
    val wordsRDD = lines.flatMap(x=>x.split("\\s+")).map(x=>(x,1))
    val wordCountRDD = wordsRDD.reduceByKey((x,y)=>x+y)
    wordCountRDD.foreach(x=>println(x))
---------输出-----------
(ee,6)
(aa,5)
(dd,2)
(zz,1)
(zks,2)
(kks,1)
(ff,1)
(bb,2)
(cc,1)

阅读全文 »

Spark RDD常用算子操作（五） 键值对聚合操作 combineByKey

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

键值对聚合操作 combineByKey

combineByKey

聚合数据一般在集中式数据比较方便，如果涉及到分布式的数据集，该如何去实现呢。这里介绍一下combineByKey, 这个是各种聚集操作的鼻祖，应该要好好了解一下,参考
scala API

简要介绍

1
2
3
def combineByKey[C](createCombiner: (V) => C,  
                    mergeValue: (C, V) => C,   
                    mergeCombiners: (C, C) => C): RD
  • createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素， combineByKey() 会使用一个叫做createCombiner() 的函数来创建那个键对应的累加器的初始值
  • mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用 mergeValue() 方法将该键的累加器对应的当前值与这个新的值进行合并
  • mergeCombiners: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。
阅读全文 »

Spark RDD常用算子操作（四） 创建键值对RDD mapToPair flatMapToPair

发表于 | 分类于 | 阅读次数：

本文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

创建键值对RDD mapToPair flatMapToPair

mapToPair

举例，在F:\sparktest\sample.txt 文件的内容如下

1
2
3
4
aa bb cc aa aa aa dd dd ee ee ee ee 
ff aa bb zks
ee kks
ee  zz zks

将每一行的第一个单词作为键，1 作为value创建pairRDD
scala版本
scala是没有mapToPair函数的，scala版本只需要map就可以了

1
2
3
4
5
6
scala> val lines = sc.textFile("F:\\sparktest\\sample.txt")

scala> val pairs = lines.map(x => (x.split("\\s+")(0), 1))

scala> pairs.collect
res0: Array[(String, Int)] = Array((aa,1), (ff,1), (ee,1), (ee,1))

java版本

1
2
3
4
5
6
7
8
JavaRDD<String> lines = sc.textFile("F:\\sparktest\\sample.txt");
//输入的是一个string的字符串，输出的是一个(String, Integer) 的map
JavaPairRDD<String, Integer> pairRDD = lines.mapToPair(new PairFunction<String, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(String s) throws Exception {
        return new Tuple2<String, Integer>(s.split("\\s+")[0], 1);
    }
});

阅读全文 »

Spark RDD常用算子操作（三） distinct，union，intersection，subtract，cartesian

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

distinct，union，intersection，subtract，cartesian

spark伪集合
尽管 RDD 本身不是严格意义上的集合，但它也支持许多数学上的集合操作，比如合并和相交操作, 下图展示了这四种操作

distinct

distinct用于去重， 我们生成的RDD可能有重复的元素，使用distinct方法可以去掉重复的元素, 不过此方法涉及到混洗，操作开销很大
scala版本

1
2
3
4
5
6
7
8
9
scala> var RDD1 = sc.parallelize(List("aa","aa","bb","cc","dd"))

scala> RDD1.collect
res3: Array[String] = Array(aa, aa, bb, cc, dd)

scala> var distinctRDD = RDD1.distinct

scala> distinctRDD.collect
res5: Array[String] = Array(aa, dd, bb, cc)

java版本

1
2
3
4
5
6
7
8
    JavaRDD<String> RDD1 = sc.parallelize(Arrays.asList("aa", "aa", "bb", "cc", "dd"));
    JavaRDD<String> distinctRDD = RDD1.distinct();
    List<String> collect = distinctRDD.collect();
    for (String str:collect) {
        System.out.print(str+", ");
    }
---------输出----------
aa, dd, bb, cc,

阅读全文 »

Spark RDD常用算子操作（二） filter,map,flatMap

发表于 | 分类于 | 阅读次数：

原文作者：翟开顺
首发：CSDN
本人仅为自己方便查阅做了摘抄，请支持原作者
原文地址：https://blog.csdn.net/t1dmzks/article/details/72077428

filter,map,flatMap

filter

举例，在F:\sparktest\sample.txt 文件的内容如下

1
2
3
4
aa bb cc aa aa aa dd dd ee ee ee ee 
ff aa bb zks
ee kks
ee  zz zks

我要将包含zks的行的内容给找出来
scala版本

1
2
3
4
5
6
    val lines = sc.textFile("F:\\sparktest\\sample.txt").filter(line=>line.contains("zks"))
    //打印内容
    lines.collect().foreach(println(_));
-------------输出------------------
ff aa bb zks
ee  zz zks

java版本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
        JavaRDD<String> lines = sc.textFile("F:\\sparktest\\sample.txt");
        JavaRDD<String> zksRDD = lines.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                return s.contains("zks");
            }
        });
        //打印内容
        List<String> zksCollect = zksRDD.collect();
        for (String str:zksCollect) {
            System.out.println(str);
        }
----------------输出-------------------
ff aa bb zks
ee  zz zks

阅读全文 »
0%