Scala开发

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
  * Created by Administrator on 2017/10/21.
  */
object TransFormation {
  def main(args: Array[String]): Unit = {
    //map()
    //filter()
    //flatMap()
    //groupByKey()
    //reduceByKey()
    //sortByKey()
    //join()
    //union()
    //intersection()
    //distinct()
    cartesian()

  }

  def cartesian(): Unit ={
    val conf = new SparkConf().setAppName("cartesian").setMaster("local")
    val sc = new SparkContext(conf)
    var list =List(1,2,3,4,5,6,7)
    var list1=List("a","b","c","d")
    sc.parallelize(list).cartesian(sc.parallelize(list1)).foreach(t=>println(t._1+"\t"+t._2))
  }

  def distinct(): Unit ={
    val conf = new SparkConf().setAppName("distinct").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List(1,1,1,2,2,3,4,3,5,6,4,5,7)
    sc.parallelize(list).distinct().foreach(println(_))
  }

  def intersection(): Unit ={
    val conf = new SparkConf().setAppName("intersection").setMaster("local")
    val sc = new SparkContext(conf)
    var list =List(1,2,3,4,5,6)
    var list1=List(4,5,6,7,8,9)
    sc.parallelize(list).intersection(sc.parallelize(list1)).foreach(println(_))
  }

  def union(): Unit ={
    val conf = new SparkConf().setAppName("union").setMaster("local")
    val sc = new SparkContext(conf)
    var list =List(1,2,3,4)
    var list1=List(5,6,7,8)
    sc.parallelize(list).union(sc.parallelize(list1)).foreach(println(_))
  }
  def join(): Unit ={
    val conf = new SparkConf().setAppName("join").setMaster("local")
    val sc = new SparkContext(conf)
    var list1=List((1,"hadoop"),(2,"spark"),(3,"hbase"))
    var list2=List((1,"had"),(2,"spa"),(3,"hba"))
    sc.parallelize(list1).join(sc.parallelize(list2)).foreach(t=>println(t._1+"\t"+t._2._1+"\t"+t._2._2))
  }
  def sortByKey(): Unit ={
    val conf = new SparkConf().setAppName("sortByKey").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List((6,"hadoop"),(8,"spark"),(10,"hbase"))
    sc.parallelize(list).sortByKey().foreach(t=>println(t._1+"\t"+t._2))
  }
  def reduceByKey(): Unit ={
    val conf = new SparkConf().setAppName("reduceByKey").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List(("hadoop",111),("spark",222),("hadoop",333),("spark",444))
    sc.parallelize(list).reduceByKey(_+_).foreach(t=>println(t._1+"\t"+t._2))

  }

  def groupByKey(): Unit ={
    val conf = new SparkConf().setAppName("groupByKey").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List((1->"docu"),(2->"idontkoow"))
    sc.parallelize(list).groupByKey().foreach(t=>println(t._1+"\t"+t._2))
  }
  def flatMap(): Unit ={
    val conf = new SparkConf().setAppName("flatMap").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List("you,jump","i,jump")
    sc.parallelize(list).flatMap(_.split(",")).foreach(println(_))
  }

  def filter(): Unit ={
    val conf = new SparkConf().setAppName("filter").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List(1,2,3,4,5,6,7,8,9,10)
    sc.parallelize(list).filter(_%2==0).foreach(println(_))
  }
  def map(): Unit ={
    val conf = new SparkConf().setAppName("map").setMaster("local")
    val sc = new SparkContext(conf)
    var list=List("hadoop","spark","hive")
    sc.parallelize(list).map("hello"+_).foreach(x=>println(x))
  }

}

Java开发

import org.apache.spark.SparkConf;
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.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Function2;
import scala.Tuple2;
import scala.reflect.runtime.SynchronizedSymbols;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Stream;

/**
 * Created by Administrator on 2017/10/21.
 */
public class TransFormationOperator {

    public static void main(String[] args) {
        //map();
        //filter();
       //flatMap();
        //groupByKey();
        //reduceByKey();
        //sortByKey();
        //join();
        //union();
        //intersection();
        //distinct();
       // cartesian();


    }
    //笛卡尔积
      public static void cartesian(){
          SparkConf conf = new SparkConf().setAppName("cartesian").setMaster("local");
          JavaSparkContext sc = new JavaSparkContext(conf);
          List<Integer> list1 = Arrays.asList(1, 2, 3, 4, 5);
          List<String> list2 = Arrays.asList("hadoop", "spark", "hive", "hbase");
          JavaRDD<Integer> list1RDD = sc.parallelize(list1);
          JavaRDD<String> list2RDD = sc.parallelize(list2);
          list1RDD.cartesian(list2RDD).foreach(new VoidFunction<Tuple2<Integer, String>>() {
              @Override
              public void call(Tuple2<Integer, String> t) throws Exception {
                  System.out.println(t._1+"<==>"+t._2);
              }
          });
      }
    //去重
    public static void distinct(){
        SparkConf conf = new SparkConf().setAppName("distinct").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list = Arrays.asList("hadoop", "hadoop", "hadoop", "spark", "spark","hbase");
        JavaRDD<String> listRDD = sc.parallelize(list);
        JavaRDD<String> distinct = listRDD.distinct();
        distinct.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }

    //求交集
    public static void intersection(){
        SparkConf conf = new SparkConf().setAppName("intersection").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        final List<String> list1 = Arrays.asList("hadoop", "spark", "hbase","hive");

        final List<String> list2 = Arrays.asList("hbase", "hive", "zookeeper");

        JavaRDD<String> list1RDD = sc.parallelize(list1);
        JavaRDD<String> list2RDD = sc.parallelize(list2);

        JavaRDD<String> intersection = list1RDD.intersection(list2RDD);
        intersection.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
    }
    //求并集
    public static void union(){
        SparkConf conf = new SparkConf().setAppName("union").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list1 = Arrays.asList("hadoop", "spark", "hbase","hive");

        List<String> list2 = Arrays.asList("hive", "sqoop", "akka");

        JavaRDD<String> list1RDD = sc.parallelize(list1);
        JavaRDD<String> list2RDD = sc.parallelize(list2);

        JavaRDD<String> union = list1RDD.union(list2RDD);
        union.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });


    }

    public  static void join(){
        SparkConf conf = new SparkConf().setAppName("join").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<Integer, String>> list1 = Arrays.asList(
                new Tuple2<Integer, String>(11, "mayun"),
                new Tuple2<Integer, String>(22, "mahuateng"),
                new Tuple2<Integer, String>(33, "zll"));

        List<Tuple2<Integer, String>> list2 = Arrays.asList(
                new Tuple2<Integer, String>(11, "alibaba"),
                new Tuple2<Integer, String>(22, "tenxun"),
                new Tuple2<Integer, String>(33, "zsjituan")
        );
        JavaPairRDD<Integer, String> list1RDD = sc.parallelizePairs(list1);
        JavaPairRDD<Integer, String> list2RDD = sc.parallelizePairs(list2);

        JavaPairRDD<Integer, Tuple2<String, String>> joinRDD = list1RDD.join(list2RDD);
        joinRDD.foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, String>>>() {
            @Override
            public void call(Tuple2<Integer, Tuple2<String, String>> t) throws Exception {
                System.out.println(t._1+"<==>"+t._2._1+"<==>"+t._2._2);
            }
        });


    }

    public static void sortByKey(){
        final SparkConf conf = new SparkConf().setAppName("sortBykey").setMaster("local");

        final JavaSparkContext sc = new JavaSparkContext(conf);

        final List<Tuple2<Integer, String>> list = Arrays.asList(
                new Tuple2<Integer, String>(77, "hadoop"),
                new Tuple2<Integer, String>(44, "spark"),
                new Tuple2<Integer, String>(55, "hive"),
                new Tuple2<Integer, String>(66, "hbase")
        );

        JavaRDD<Tuple2<Integer, String>> listRDD = sc.parallelize(list);
        listRDD.mapToPair(new PairFunction<Tuple2<Integer,String>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<Integer, String> t) throws Exception {
                return new Tuple2<Integer,String>(t._1,t._2);
            }
        }).sortByKey().foreach(new VoidFunction<Tuple2<Integer, String>>() {
            @Override
            public void call(Tuple2<Integer, String> t) throws Exception {
                System.out.println(t._2+"<--->"+t._1);
            }
        });
        sc.stop();
    }

    public static void reduceByKey(){
        final SparkConf conf = new SparkConf().setAppName("reduceByKey").setMaster("local");
        final JavaSparkContext sc = new JavaSparkContext(conf);

        final List<Tuple2<String, Integer>> list = Arrays.asList(
                new Tuple2<String, Integer>("hadoop", 90),
                new Tuple2<String, Integer>("spark", 80),
                new Tuple2<String, Integer>("hbase", 85),
                new Tuple2<String, Integer>("hive", 82)
        );

        final JavaRDD<Tuple2<String, Integer>> listRDD = sc.parallelize(list);
        listRDD.mapToPair(new PairFunction<Tuple2<String,Integer>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<String, Integer> t) throws Exception {
                return new Tuple2<String,Integer>(t._1,t._2);
            }
        }).reduceByKey(new org.apache.spark.api.java.function.Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1+i2;
            }
        }).foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> t) throws Exception {
                System.out.println(t._1+"==>"+t._2);
            }
        });
        sc.stop();
    }

    public static void groupByKey(){
        SparkConf conf = new SparkConf().setAppName("groupByKey").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Tuple2<String, String>> list = Arrays.asList(
                new Tuple2<String, String>("hadoop", "had"),
                new Tuple2<String, String>("spark", "spa"),
                new Tuple2<String, String>("hbase", "hba"),
                new Tuple2<String, String>("hive", "hiv")
        );
        JavaRDD<Tuple2<String, String>> listRDD = sc.parallelize(list);
        listRDD.groupBy(new org.apache.spark.api.java.function.Function<Tuple2<String,String>, String>() {
            @Override
            public String call(Tuple2<String, String> t) throws Exception {
                return t._1;
            }
        }).foreach(new VoidFunction<Tuple2<String, Iterable<Tuple2<String, String>>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Tuple2<String, String>>> t) throws Exception {
                String menpai=t._1;
                Iterator<Tuple2<String, String>> iterator = t._2.iterator();
                System.out.println(menpai);
                while(iterator.hasNext()){
                    Tuple2<String, String> ren = iterator.next();
                    //System.out.println(ren.toString());
                    System.out.println( ren._2);
                }

            }
        });
        sc.stop();
    }
    public static void flatMap(){
        SparkConf conf = new SparkConf().setAppName("flatMap").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list = Arrays.asList("you,jump", "i,jump");
        JavaRDD<String> listRDD = sc.parallelize(list);
        listRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(",")).iterator();
            }
        }).foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
        sc.stop();
    }
    //过滤奇数偶数
    public static void filter(){
        final SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");
        final JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> listRDD = sc.parallelize(list);
        listRDD.filter(new org.apache.spark.api.java.function.Function<Integer, Boolean>() {
            @Override
            public Boolean call(Integer i) throws Exception {
                return i % 2==0;
            }
        }).foreach(new VoidFunction<Integer>() {
            @Override
            public void call(Integer t) throws Exception {
                System.out.println(t+"\t");
            }
        });
        sc.stop();
    }
    public static void map() {
        final SparkConf conf = new SparkConf().setAppName("map").setMaster("local");
        final JavaSparkContext sc = new JavaSparkContext(conf);
        List<String> list = new ArrayList<String>();
        list.add("hadoop");
        list.add("spark");
        list.add("hbase");
        JavaRDD<String> listRDD = sc.parallelize(list);
        listRDD.map(new org.apache.spark.api.java.function.Function<String, Tuple2<String,Integer>>() {

            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String,Integer>(s,1993);
            }
        }).foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> t) throws Exception {
                System.out.println(t._1+"\t"+t._2);
            }
        });
        sc.stop();

    }
}