Java Spark Dataset MapFunction - 任务在没有任何类引用的情况下不可序列化

Posted 2023-03-31

【中文标题】Java Spark Dataset MapFunction - 任务在没有任何类引用的情况下不可序列化

【英文标题】：Java Spark Dataset MapFunction - Task not serializable without any reference to class

【发布时间】：2021-02-16 22:31:39

【问题描述】：

我有一个将 csv 数据读入 Spark 的Dataset 的类。如果我只是简单地阅读并返回data，一切都会正常工作。

但是，如果我在从函数返回之前将MapFunction 应用于data，我会得到

Exception in thread "main" org.apache.spark.SparkException: Task not serializable

Caused by: java.io.NotSerializableException: com.Workflow.

我知道 Spark 的工作原理以及它需要序列化对象以进行分布式处理，但是，我没有在我的映射逻辑中使用对 Workflow 类的任何引用。我没有在我的映射逻辑中调用任何Workflow 类函数。那么为什么 Spark 会尝试序列化 Workflow 类呢？任何帮助将不胜感激。

public class Workflow 

    private final SparkSession spark;   

    public Dataset<Row> readData()
        final StructType schema = new StructType()
            .add("text", "string", false)
            .add("category", "string", false);

        Dataset<Row> data = spark.read()
            .schema(schema)
            .csv(dataPath);

        /* 
         * works fine till here if I call
         * return data;
         */

        Dataset<Row> cleanedData = data.map(new MapFunction<Row, Row>() 
            public Row call(Row row)
                /* some mapping logic */
                return row;
            
        , RowEncoder.apply(schema));

        cleanedData.printSchema();
        /* .... ERROR .... */
        cleanedData.show();

        return cleanedData;
    

【问题讨论】：

看看这是否有帮助：***.com/questions/40818001/…

【参考方案1】：

匿名内部类具有对封闭类的隐藏/隐式引用。使用 Lambda 表达式或使用 Roma Anankin 的解决方案

【参考方案2】：

你可以让 Workflow 实现 Serializeble 和 SparkSession 作为@transient

【讨论】：

虽然这可能有效，但它不能回答问题，即为什么当我没有在 map 函数中引用它时，Spark 会尝试序列化 Workflow。