在 java 中, RDD 分为 javaRDDs 和 javaPairRDDs. 下面分两大类来进行.
都必须要进行的一步.
- SparkConf conf = new SparkConf().setMaster("local").setAppName("test");
- JavaSparkContext sc = new JavaSparkContext(conf);
一. javaRDDs
- String[] ayys = {"a","b","c"};
- List<String> strings = Arrays.asList(ayys);
- JavaRDD<String> rdd1 = sc.parallelize(strings);
- strings.add("d");
- JavaRDD<String> rdd2 = sc.parallelize(strings);
- JavaRDD<Tuple2<String, Integer>> parallelize = sc.parallelize(Arrays.asList(
- new Tuple2<String, Integer>("asd", 11),
- new Tuple2<String, Integer>("asd", 11),
- new Tuple2<String, Integer>("asd", 11)
- ));
- rdd1.map(new Function<String, String>() {
- public String call(String s) throws Exception {
- return s.replace("a","qqq");
- }
- }).foreach(new VoidFunction<String>() {
- public void call(String s) throws Exception {
- System.out.println(s);
- }
- });
- List<String> a = rdd1.filter(new Function<String, Boolean>() {
- public Boolean call(String s) throws Exception {
- return s.contains("a");
- }
- }).collect();
- System.out.println(a);
- JavaRDD<String> rdd22 = rdd1.flatMap(new FlatMapFunction<String, String>() {
- public Iterable<String> call(String s) throws Exception {
- return Arrays.asList(s.split(" "));
- }
- });
- JavaPairRDD<String, Integer> rdd4 = rdd2.mapToPair(new PairFunction<String, String, Integer>() {
- public Tuple2<String, Integer> call(String s) throws Exception {
- return new Tuple2<String, Integer>(s, 1);
- }
- });
- JavaRDD<String> rdd11 = rdd2.mapPartitions(new FlatMapFunction<Iterator<String>, String>() {
- public Iterable<String> call(Iterator<String> stringIterator) throws Exception {
- ArrayList<String> strings = new ArrayList<String>();
- while (stringIterator.hasNext()){
- strings.add(stringIterator.next());
- }
- return strings;
- }
- });
- JavaRDD<String> stringJavaRDD = rdd1.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {
- public Iterator<String> call(Integer integer, Iterator<String> stringIterator) throws Exception {
- ArrayList<String> strings = new ArrayList<String>();
- while (stringIterator.hasNext()){
- strings.add(stringIterator.next());
- }
- return strings.iterator();
- }
- },false);
- JavaRDD<String> sample = rdd1.sample(false, 0.3);
- JavaRDD<String> union = rdd1.union(rdd2);
- JavaRDD<String> intersection = rdd1.intersection(rdd2);
- JavaRDD<String> distinct = rdd1.distinct();
二. JavaPairRDDs.
- JavaPairRDD<String, Integer> rdd1 = sc.parallelizePairs(Arrays.asList(
- new Tuple2<String, Integer>("asd", 111),
- new Tuple2<String, Integer>("asd", 111),
- new Tuple2<String, Integer>("asd", 111)
- ));
- JavaPairRDD<String, Integer> rdd2 = sc.parallelizePairs(Arrays.asList(
- new Tuple2<String, Integer>("sdfsd", 222),
- new Tuple2<String, Integer>("sdfsd", 222),
- new Tuple2<String, Integer>("sdfsd", 222)
- ));
- JavaPairRDD<String, Iterable<Integer>> stringIterableJavaPairRDD = rdd1.groupByKey();
- JavaPairRDD<String, Integer> rdd = rdd1.reduceByKey(new Function2<Integer, Integer, Integer>() {
- public Integer call(Integer integer, Integer integer2) throws Exception {
- return integer + integer2;
- }
- });
- JavaPairRDD<String, Integer> rdd3 = rdd1.aggregateByKey(0, new Function2<Integer, Integer, Integer>() {
- public Integer call(Integer integer, Integer integer2) throws Exception {
- return max(integer,integer2);
- }
- }, new Function2<Integer, Integer, Integer>() {
- public Integer call(Integer integer, Integer integer2) throws Exception {
- return integer + integer2;
- }
- });
- JavaPairRDD<String, Integer> rdd111 = rdd1.sortByKey();
- JavaPairRDD<String, Tuple2<Integer, Integer>> join = rdd1.join(rdd2);
- JavaPairRDD<String, Tuple2<Integer, Optional<Integer>>> stringTuple2JavaPairRDD = rdd1.leftOuterJoin(rdd2);
- JavaPairRDD<String, Tuple2<Optional<Integer>, Integer>> stringTuple2JavaPairRDD1 = rdd1.rightOuterJoin(rdd2);
- JavaPairRDD<String, Tuple2<Optional<Integer>, Optional<Integer>>> stringTuple2JavaPairRDD2 = rdd1.fullOuterJoin(rdd2);
- JavaPairRDD<String, Tuple2<Iterable<Integer>, Iterable<Integer>>> cogroup = rdd1.cogroup(rdd2);
- JavaPairRDD<String, Integer> coalesce = rdd1.coalesce(3, false);
- JavaPairRDD<String, Integer> repartition = rdd1.repartition(3);
- JavaPairRDD<String, Integer> rdd5 = rdd1.repartitionAndSortWithinPartitions(new HashPartitioner(2));
- JavaPairRDD<Tuple2<String, Integer>, Tuple2<String, Integer>> cartesian = rdd1.cartesian(rdd2);
- JavaRDD<String> pipe = rdd1.pipe("");
最后都要加上
sc.stop();
aggregateByKey 算子详解 https://www.cnblogs.com/LHWorldBlog/p/8215529.html
repartitionAndSortWithinPartitions 算子详解
来源: http://www.bubuko.com/infodetail-3008193.html