Spark RDD操作之Map系算子

落日映苍穹つ 2022-01-16 21:29 219阅读 0赞

[为什么80%的码农都做不了架构师？>>> ][80_]  ![hot3.png][]

本篇博客将介绍Spark RDD的Map系算子的基本用法。

1、map

map将RDD的元素一个个传入call方法，经过call方法的计算之后，逐个返回，生成新的RDD，计算之后，记录数不会缩减。示例代码，将每个数字加10之后再打印出来， 代码如下

import java.util.Arrays;
    
    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 Map {
    	public static void main(String[] args) {
    		SparkConf conf = new SparkConf().setAppName("spark map").setMaster("local[*]");
    		JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
    		JavaRDD<Integer> listRDD = javaSparkContext.parallelize(Arrays.asList(1, 2, 3, 4));
    
    		JavaRDD<Integer> numRDD = listRDD.map(new Function<Integer, Integer>() {
    			@Override
    			public Integer call(Integer num) throws Exception {
    				return num + 10;
    			}
    		});
    		numRDD.foreach(new VoidFunction<Integer>() {
    			@Override
    			public void call(Integer num) throws Exception {
    				System.out.println(num);
    			}
    		});
    	}
    
    }

执行结果：

![086924c57ebfca7397f7a513d7986613328.jpg][]

2、flatMap

flatMap和map的处理方式一样，都是把原RDD的元素逐个传入进行计算，但是与之不同的是，flatMap返回值是一个Iterator，也就是会一生多，超生

import java.util.Arrays;
    import java.util.Iterator;
    
    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.FlatMapFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    
    public class FlatMap {
    	public static void main(String[] args) {
    		SparkConf conf = new SparkConf().setAppName("spark map").setMaster("local[*]");
    		JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
    		JavaRDD<String> listRDD = javaSparkContext
    				.parallelize(Arrays.asList("hello wold", "hello java", "hello spark"));
    		JavaRDD<String> rdd = listRDD.flatMap(new FlatMapFunction<String, String>() {
    			private static final long serialVersionUID = 1L;
    
    			@Override
    			public Iterator<String> call(String input) throws Exception {
    				return Arrays.asList(input.split(" ")).iterator();
    			}
    		});
    		rdd.foreach(new VoidFunction<String>() {
    			private static final long serialVersionUID = 1L;
    			@Override
    			public void call(String num) throws Exception {
    				System.out.println(num);
    			}
    		});
    	}
    
    }

执行结果：

![49fd6062f5fc6e5f1220db2311fad56f5f3.jpg][]

3、mapPartitions

mapPartitions一次性将整个分区的数据传入函数进行计算，适用于一次性聚会整个分区的场景

public class MapPartitions {
    	public static void main(String[] args) {
    		SparkConf conf = new SparkConf().setAppName("spark map").setMaster("local[*]");
    		JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
    		JavaRDD<String> listRDD = javaSparkContext.parallelize(Arrays.asList("hello", "java", "wold", "spark"), 2);
    
    		/**
    		 * mapPartitions回调的接口也是FlatMapFunction，FlatMapFunction的第一个泛型是Iterator表示传入的数据，
    		 * 第二个泛型表示返回数据的类型
    		 * 
    		 * mapPartitions传入FlatMapFunction接口处理的数据是一个分区的数据，所以，如果一个分区数据过大，会导致内存溢出
    		 * 
    		 */
    		JavaRDD<String> javaRDD = listRDD.mapPartitions(new FlatMapFunction<Iterator<String>, String>() {
    			int i = 0;
    
    			@Override
    			public Iterator<String> call(Iterator<String> input) throws Exception {
    				List<String> list = new ArrayList<String>();
    				while (input.hasNext()) {
    					list.add(input.next() + i);
    					++i;
    				}
    				return list.iterator();
    			}
    		});
    
    		javaRDD.foreach(new VoidFunction<String>() {
    			@Override
    			public void call(String t) throws Exception {
    				System.out.println(t);
    			}
    		});
    	}
    
    }

运行结果:

![5c73664a2ecd2696d79b73020b776f5b471.jpg][]

上面的运算结果，后面的尾标只有0和1，说明FlatMapFunction被调用了两次，与MapPartitions的功能吻合。

4、mapPartitionsWithIndex

mapPartitionsWithIndex和mapPartitions一样，一次性传入整个分区的数据进行处理，但是不同的是，这里会传入分区编号进来

public class mapPartitionsWithIndex {
    	public static void main(String[] args) {
    		SparkConf conf = new SparkConf().setAppName("spark map").setMaster("local[*]");
    		JavaSparkContext javaSparkContext = new JavaSparkContext(conf);
    		JavaRDD<String> listRDD = javaSparkContext.parallelize(Arrays.asList("hello", "java", "wold", "spark"), 2);
    
    		/**
    		 *和mapPartitions一样，一次性传入整个分区的数据进行处理，但是不同的是，这里会传入分区编号进来
    		 * 
    		 */
    		JavaRDD<String> javaRDD = listRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {
    
    			@Override
    			public Iterator<String> call(Integer v1, Iterator<String> v2) throws Exception {
    				List<String> list = new ArrayList<String>();
    				while (v2.hasNext()) {
    					list.add(v2.next() + "====分区编号："+v1);
    				}
    				return list.iterator();
    			}
    		
    		},true);
    
    		javaRDD.foreach(new VoidFunction<String>() {
    			@Override
    			public void call(String t) throws Exception {
    				System.out.println(t);
    			}
    		});
    	}
    
    }

执行结果：

![22857cf67c18f88b382349a4081deafa0de.jpg][]

转载于:https://my.oschina.net/u/1778239/blog/3019466

[80_]: https://my.oschina.net/u/2663968/blog/3051541
[hot3.png]: /images/20220117/096cadfa224d4168a9b038a706f5cef1.png
[086924c57ebfca7397f7a513d7986613328.jpg]: /images/20220117/88d76465477d4616a8b09151f46d6673.png
[49fd6062f5fc6e5f1220db2311fad56f5f3.jpg]: /images/20220117/c131159bff7742af9f6b4861c1b7a47a.png
[5c73664a2ecd2696d79b73020b776f5b471.jpg]: /images/20220117/3229d1a48c18480a8b2b0bce57c052b5.png
[22857cf67c18f88b382349a4081deafa0de.jpg]: /images/20220117/fb4dfff29d6743088d352718cd256f4a.png