spark—键值对操作

2018-11-07 本文已影响6人 Sophie12138

1.JavaPairRDD背景
键值对 RDD 通常用来进行聚合计算。先通过一些初始 ETL（抽取、转化、装载）操作来将数据转化为键值对形式。键值对 RDD 提供了一些新的操作接口

让用户控制键值对 RDD 在各节点上分布情况的高级特性：分区。
使用可控的分区方式把常被一起访问的数据放到同一个节点上，可以大大减少应用的通信开销。这会带来明显的性能提升。

Spark 为包含键值对类型的 RDD 提供了一些专有的操作，称为 pair RDD。它们提供了并行操作各个键或跨节点重新进行数据分组的操作接口。

通常从一个 RDD 中提取某些字段（例如代表事件时间、用户 ID 或者其他标识符的字段），使用这些字段作为 pair RDD 操作中的键。

2.基本操作

filter：对pair RDD的元素按照某种规则进行过滤，去掉不需要的pair RDD
reduceByKey：对pairRDD中的元素中所有相同key对应的元素进行统计，下面是单词个数统计的例子

rdd.flatMap(line -> Arrays.asList(line.split("")).iterator())
    .mapToPair(word -> new Tuple2<String, Integer>(word, 1))
    .reduceByKey((s1, s2) -> s1 + s2)// 将相同的key进行reduce，并将value相加

combineByKey：groupByKey、reduceByKey都是由combineByKey实现的。该函数用于将RDD[K,V]转换成RDD[K,C],这里的V类型和C类型可以相同也可以不同。此函数有三个参数必传：
createCombiner：组合器函数，用于将V类型转换成C类型，输入参数为RDD[K,V]中的V,输出为C
mergeValue：合并值函数，将一个C类型和一个V类型值合并成一个C类型，输入参数为(C,V)，输出为C
mergeCombiners：合并组合器函数，用于将两个C类型值合并成一个C类型，输入参数为(C,C)，输出为C
numPartitions：结果RDD分区数，不传则默认保持原有的分区数
partitioner：分区函数，不传则默认为HashPartitioner
mapSideCombine：是否需要在Map端进行combine操作，类似于MapReduce中的combine，不传则默认为true

combineByKey()的实现是一边进行aggregate，一边进行compute() 的基础操作。假设一组具有相同 K 的 <K, V> records 正在一个个流向 combineByKey()，createCombiner将第一个 record 的 value 初始化为 c （比如，c = value），然后从第二个 record 开始，来一个 record 就使用 mergeValue(c, record.value) 来更新 c，如果想对这些 records 的所有 values 做 sum，那么使用 c = c + record.value。等到 records 全部被 mergeValue()，得到结果 c。假设还有一组 records（key 与前面那组的 key 均相同）一个个到来，combineByKey() 使用前面的方法不断计算得到 c'。现在如果要求这两组 records 总的 combineByKey() 后的结果，那么可以使用 final c = mergeCombiners(c, c') 来计算；然后依据partitioner进行不同分区合并。

List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7, 1, 2);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
//转化为pairRDD
JavaPairRDD<Integer,Integer> javaPairRDD = javaRDD.mapToPair(
                            new PairFunction<Integer, Integer, Integer>() {    
    @Override    
    public Tuple2<Integer, Integer> call(Integer integer) throws Exception {  
      return new Tuple2<Integer, Integer>(integer,1);   
  }
});

JavaPairRDD<Integer,String> combineByKeyRDD = javaPairRDD.combineByKey(
  new Function<Integer, String>() {    
    @Override    
    public String call(Integer v1) throws Exception {  
      return v1 + " :createCombiner: ";    
  }}, 
  new Function2<String, Integer, String>() {    
    @Override    
    public String call(String v1, Integer v2) throws Exception {        
      return v1 + " :mergeValue: " + v2;    
  }}, 
  new Function2<String, String, String>() {    
    @Override    
    public String call(String v1, String v2) throws Exception {        
      return v1 + " :mergeCombiners: " + v2;    
  }});

设置分区个数：每个RDD都有固定数目的分区，在建立RDD时，如果不指定分区个数的话，系统会根据集群大小推断出一个有意义的默认值。

3.数据分组

groupByKey():对pair RDD中相同key对应的元素进行合并，形成一个元素列表作为value，key值保持不变。

4.数据连接

join：对于2个pair RDD中都含有的key，设其value为v1,v2，返回一个pair RDD，其中的元素为(key, (v1,v2))。
leftOuterJoin与rightOuterJoin：允许key值有缺失的连接，分别允许右边与左边的值有缺失，对应为None

5.数据排序

sortByKey：按照key进行升序或者降序排列，可以指定比较key时采用的函数，即不是比较key1与key2，而是比较f(key1)与f(key2)，但是f(key)的返回值类型需要与key相同。

6.pair RDD的行动操作

countByKey：对每个key对应的元素个数进行计数
collectAsMap：将结果以映射表的形式返回，便于查询
lookup：返回给定键对应的所有值，为list的形式

spark—键值对操作

猜你喜欢

热点阅读