org.apache.spark.SparkException: Task not serializable问题

刺骨的言语ヽ痛彻心扉 2022-05-26 11:38 127阅读 0赞

由于Spark程序中的map、filter等算子内部引用了类成员函数或变量导致该类所有成员都需要支持序列化，又由于该类某些成员变量不支持序列化，最终引发Task无法序列化问题。相反地，对类中那些不支持序列化问题的成员变量标注后，使得整个类能够正常序列化，最终消除Task未序列化问题。

public class SparkRDDS {
    	 SparkConf conf;//定义配置信息对象
    	 JavaSparkContext jsc ;//声明spark上下文
    	@Before
    	public void start(){
    		 conf = new SparkConf().setMaster("local[1]").setAppName("sparkRDDs");
    		 jsc = new JavaSparkContext(conf);
    	}
    	@After
    	public void end(){
    		jsc.close();
    	}
    	/**
    	 * map算子
    	 * 对每一个元素执行操作
    	 */
    	@Test
    	public void mapRDDTest(){
    		//准备数据
    		List<Integer> list = Arrays.asList(1,2,3,4,5);
    		//数据并行化
    		JavaRDD<Integer> dataRDD = jsc.parallelize(list);
    		//执行map算子操作
    		JavaRDD<Integer> mapRDD = dataRDD.map(new Function<Integer, Integer>() {
    			private static final long serialVersionUID = 1L;
    			@Override
    			public Integer call(Integer ele) throws Exception {
    				int num = ele*10;
    				return num;
    			}
    		});
    		mapRDD.foreach(new VoidFunction<Integer>() {
    			private static final long serialVersionUID = 1L;
    
    			@Override
    			public void call(Integer val) throws Exception {
    				System.out.println(val);
    			}
    		});
    	}
    	
    }

由于依赖了当前类的成员变量，所以导致当前类全部需要序列化，由于当前类某些字段未做好序列化，导致出错。实际情况与分析的原因一致，运行过程中出现错误，如下所示。分析下面的错误报告得到错误是由于SparkConf和JavaSparkContext引起的。

![201805011437422][]

将此类实现Serializable可序列化，将不需要序列化的的成员变量使用关键字transient标注。

public class SparkRDDS implements Serializable {
    	transient SparkConf conf;//定义配置信息对象
    	transient JavaSparkContext jsc ;//声明spark上下文
    	@Before
    	public void start(){
    		 conf = new SparkConf().setMaster("local[1]").setAppName("sparkRDDs");
    		 jsc = new JavaSparkContext(conf);
    	}
    	@After
    	public void end(){
    		jsc.close();
    	}
    	/**
    	 * map算子
    	 * 对每一个元素执行操作
    	 */
    	@Test
    	public void mapRDDTest(){
    		//准备数据
    		List<Integer> list = Arrays.asList(1,2,3,4,5);
    		//数据并行化
    		JavaRDD<Integer> dataRDD = jsc.parallelize(list);
    		//执行map算子操作
    		JavaRDD<Integer> mapRDD = dataRDD.map(new Function<Integer, Integer>() {
    			private static final long serialVersionUID = 1L;
    			@Override
    			public Integer call(Integer ele) throws Exception {
    				int num = ele*10;
    				return num;
    			}
    		});
    		mapRDD.foreach(new VoidFunction<Integer>() {
    			private static final long serialVersionUID = 1L;
    
    			@Override
    			public void call(Integer val) throws Exception {
    				System.out.println(val);
    			}
    		});
    	}	
    }

再次执行时，程序即可正常执行。

[201805011437422]: /images/20220526/46c40150de6a4e3abcf29c87e0711bf6.png