Spark SQL UDF任务不可序列化(Spark SQL UDF Task not serializable)

 卡桑德拉和DataStax社区，我有一个问题，我希望有人明智可以帮助我。 
 
 我们正在将我们的分析代码从Hadoop迁移到运行在Cassandra之上的Spark（通过DataStax Enterprise）。 DSE 4.7在生产，但在开发中4.8。 
 
 Java 7正在开发中，或者Java 7/8正在开发中。 
 
 我们需要一些DataFrame转换，并且我们认为通过Spark SQLContext对内存DataFrame使用UDF可以完成这项工作。 主要是： 
 
 
 
 我们数据的每个文本值都带有前缀，后缀为“。 即“一些数据”这是非常烦人的，所以我们想要清理每个这些。 
 
 
 我们想添加一个列，其中包含由许多其他列组成的散列键。 
 
 
 我们的代码如下。 如果没有在sqlContext中包含UDF调用，这会运行良好，但只要添加了，我们就会得到“任务不可序列化”错误 
 
 线程“main”中的异常org.apache.spark.SparkException：任务不可序列化 
 
 我曾尝试将“implements Serializable”作为此类（以及许多其他类）的基类，它将错误类更改为链上的下一个类，但是，只要Exception类上的失败不可串行化......可能意味着我正走向错误的方向。 
 
 我也尝试将UDFs作为lambda进行实现，这也导致了相同的错误。 
 
 如果有人能够指出我做错了什么，我将不胜感激！ 
 
public class entities implements Serializable{
    private spark_context m_spx = null;
    private DataFrame m_entities = null;
    private String m_timekey = null;

    public entities(spark_context _spx, String _timekey){
        m_spx = _spx;
        m_timekey = _timekey;
    }


    public DataFrame get_dimension(){
        if(m_entities == null) {

            DataFrame df = m_spx.get_flat_data(m_timekey).select("event", "url");

            //UDF to generate hashed ids
            UDF2 get_hashed_id = new UDF2<String, String, String>() {
                public String call(String o, String o2) throws Exception {
                    return o.concat(o2);
                }
            };


            //UDF to clean the " from strings
            UDF1 clean_string = new UDF1<String, String>() {
                public String call(String o) throws Exception {
                    return o.replace("\"","");
                }
            };


            //Get the Spark SQL Context from SC.
            SQLContext sqlContext = new SQLContext(m_spx.sc());


            //Register the UDFs
            sqlContext.udf().register("getid", get_hashed_id, DataTypes.StringType);
            sqlContext.udf().register("clean_string", clean_string, DataTypes.StringType);


            //Register the DF as a table.
            sqlContext.registerDataFrameAsTable(df, "entities");
            m_entities = sqlContext.sql("SELECT getid(event, url) as event_key, clean_string(event) as event_cleaned, clean_string(url) as url_cleaned FROM entities");
        }

        return m_entities;
    }
}

Cassandra & DataStax community, I have a question that I'm hoping someone wise can help me with.
 
We are migrating our analtics code from Hadoop to Spark running on top of Cassandra (via DataStax Enterprise). DSE 4.7 in production, but 4.8 in development. 
 
Java 7 in production, either Java 7/8 in development. 
 
There are a couple of DataFrame transformations we need and we think that writing a UDF used via the Spark SQLContext against an in memory DataFrame will do the job. The primary of these are: 
 
 
 
every single text value of our data is prefixed and postfixed with “. i.e. “some data” This is very annoying and so we would like to clean each of these.
 
 
we would like to add a column that contains a hashed key made up from a number of the other columns.
 
 
Our code is below. This runs well without the inclusion of the UDF calls in the sqlContext but as soon as they are added we are getting “Task is not Serializable” error
 
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
 
I have tried putting “implements Serializable” as the base class of this (and many other classes), which changes the error class to the next one up the chain, however this gets as far as failing on the Exception class is not serializable… which probably means I am heading in the wrong direction. 
 
I have also tried implemeting the UDFs as lambda’s and that also results in the same error.
 
If anyone could point out what I am doing wrong it would be much appreciated!
 
public class entities implements Serializable{
    private spark_context m_spx = null;
    private DataFrame m_entities = null;
    private String m_timekey = null;

    public entities(spark_context _spx, String _timekey){
        m_spx = _spx;
        m_timekey = _timekey;
    }


    public DataFrame get_dimension(){
        if(m_entities == null) {

            DataFrame df = m_spx.get_flat_data(m_timekey).select("event", "url");

            //UDF to generate hashed ids
            UDF2 get_hashed_id = new UDF2<String, String, String>() {
                public String call(String o, String o2) throws Exception {
                    return o.concat(o2);
                }
            };


            //UDF to clean the " from strings
            UDF1 clean_string = new UDF1<String, String>() {
                public String call(String o) throws Exception {
                    return o.replace("\"","");
                }
            };


            //Get the Spark SQL Context from SC.
            SQLContext sqlContext = new SQLContext(m_spx.sc());


            //Register the UDFs
            sqlContext.udf().register("getid", get_hashed_id, DataTypes.StringType);
            sqlContext.udf().register("clean_string", clean_string, DataTypes.StringType);


            //Register the DF as a table.
            sqlContext.registerDataFrameAsTable(df, "entities");
            m_entities = sqlContext.sql("SELECT getid(event, url) as event_key, clean_string(event) as event_cleaned, clean_string(url) as url_cleaned FROM entities");
        }

        return m_entities;
    }
}

原文：https://stackoverflow.com/questions/36176011