Spark-java相关算子

﹏ヽ暗。殇╰゛Y 2021-12-01 14:58 348阅读 0赞

# Spark-java相关算子练习 #

## Transformations算子 ##

package com.cs.java.spark;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.VoidFunction;
    
    public class Day01 {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("day01");
        conf.setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("Error");
        
    	。。。
    	。。。
    	
    
    	}
    }

### filter-过滤 ###

JavaRDD<String> lines = sc.textFile("./Test_spark/data/words2");
    //filter过滤
    JavaRDD<String> result = lines.filter(new Function<String, Boolean>() {
        @Override
        public Boolean call(String line) throws Exception {
            return "hello spark".equals(line);
        }
    });
    result.foreach(new VoidFunction<String>() {
        @Override
        public void call(String s) throws Exception {
            System.out.println(s);
        }
    });
    sc.stop();

### count-计数 ###

//count计数
    final long count = result.count();
    System.out.println(count);
    sc.stop();

### map-1:1进出一条数据 单个（个，个） ###

JavaRDD<String> lines = sc.textFile("./Test_spark/data/words2");
    JavaRDD<String> map = lines.map(new Function<String, String>() {
        @Override
        public String call(String line) throws Exception {
            return line + "#";
        }
    });
    map.foreach(new VoidFunction<String>() {
        @Override
        public void call(String s) throws Exception {
            System.out.println(s);
        }
    });
    sc.stop();

### mapToPair-进出多条数据 行（行，行） ###

JavaPairRDD<String, String> result = lines.mapToPair(new PairFunction<String, String, String>() {
        @Override
        public Tuple2<String, String> call(String s) throws Exception {
            return new Tuple2<String, String>(s,s+"**");
        }
    });
    result.foreach(new VoidFunction<Tuple2<String, String>>() {
        @Override
        public void call(Tuple2<String, String> stringStringTuple2) throws Exception {
            System.out.println(stringStringTuple2);
        }
    });
    sc.stop();

### flatMap-切分 ###

### reduceByKey-运算 ###

lines.flatMap(new FlatMapFunction<String, String>() {
        @Override
        public Iterator<String> call(String s) throws Exception {
            List<String> list = Arrays.asList(s.split(" "));
            return list.iterator();
        }
    }).mapToPair(new PairFunction<String, String,Integer>() {
        @Override
        public Tuple2<String, Integer> call(String s) throws Exception {
            return new Tuple2<String, Integer>(s,1);
        }
    }).reduceByKey(new Function2<Integer, Integer, Integer>() {
        @Override
        public Integer call(Integer v1, Integer v2) throws Exception {
            return v1+v2;
        }
    }).foreach(new VoidFunction<Tuple2<String, Integer>>() {
        @Override
        public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
            System.out.println(stringIntegerTuple2);
        }
    });

## Action算子 ##

### collect-遍历数据 ###

List<String> collect = lines.collect();
    for(String one : collect){
        System.out.println(one);
    }

### count-计数 ###

long count = lines.count();
    System.out.println(count);

### first-取第一条 ###

String first = lines.first();
    System.out.println(first);

### take-取前n条 ###

List<String> take = lines.take(3);
    System.out.println(take);

### sample-取样 ###

（有无放回，比例，取定值）

tips:比例不要大于1，否则数据会乱

JavaRDD<String> sample = lines.sample(true, 0.5);
    sample.foreach(new VoidFunction<String>() {
        @Override
        public void call(String s) throws Exception {
            System.out.println(s);
        }
    });

## 第二阶段join相关算子 ##

准备：

package com.cs.java.spark;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.Optional;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.Arrays;
    import java.util.List;
    
    public class Day02 {
        public static void main(String[] args) {
            SparkConf conf = new SparkConf();
            conf.setMaster("local");
            conf.setAppName("day02");
            JavaSparkContext sc = new JavaSparkContext(conf);
            sc.setLogLevel("Error");
            
            ...
            
            }
      }

### parallelizePairs-将数据转换成KV格式RDD ###

/**
             * parallelizePairs-将数据转换成KV格式RDD
             */
            JavaPairRDD<String,Integer> nameRDD = sc.parallelizePairs(Arrays.asList(
                    new Tuple2<String,Integer>("zhangsan", 18), new Tuple2<String,Integer>("lisi", 19), new Tuple2<String,Integer>("wangwu", 20)));
            JavaPairRDD<String,Integer> soreRDD = sc.parallelizePairs(Arrays.asList(
                    new Tuple2<String,Integer>("zhangsan", 100), new Tuple2<String,Integer>("lisi", 200), new Tuple2<String,Integer>("maliu", 400)));

### join-(kv)(k,w)=>(k,(v,w)) ###

打印公共的

JavaPairRDD<String, Tuple2<Integer, Integer>> join = nameRDD.join(soreRDD);
            join.foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, Integer>>>() {
                @Override
                public void call(Tuple2<String, Tuple2<Integer, Integer>> stringTuple2Tuple2) throws Exception {
                    System.out.println(stringTuple2Tuple2);
                }
            });

### leftOuterJoin-偏左合并，注意判是否空 ###

#### 空值会报错，需要isPresent方法判断是否有值 ####

#### c.orElse(xxxx)-给空Optionla类型数据赋值 ####

/**
             * leftOuterJoin
             * 空值会报错，需要isPresent方法判断是否有值
             * 但是我的代码没有报错的啊。。。                      ???
             * c.orElse(xxxx)-给空Optionla类型数据赋值
             */
            nameRDD.leftOuterJoin(soreRDD).foreach(new VoidFunction<Tuple2<String, Tuple2<Integer, Optional<Integer>>>>() {
                @Override
                public void call(Tuple2<String, Tuple2<Integer, Optional<Integer>>> tp) throws Exception {
    //                System.out.println(tp);
                    String a = tp._1;
                    Integer b = tp._2._1;
                    Optional c = tp._2._2;
                    if(c.isPresent()){
                        System.out.println(a+"---"+b+"---"+c);
                    }else{
                        System.out.println(a+" error");
                    }
    
                }
            });