Spark ML Transformer 中的 NullPointerException

Question

我很久没有使用 spark，所以这可能是一个新手错误，但我知道我在 java 中的方法知道如何修复 NullPointerException。但是，下面的错误让我很困惑。

设置：我有 HDFS 上的数据，我使用 Spark 读取这些数据。数据存储为 HIVE 表，因此我将使用 HiveContext。我正在使用 Spark 1.6.0 和 MLLib_2.10。 Java 是 1.7。

目标：我正在读取我想要“分类”的表格中的文本。我已经离线创建了一组（n-gram，重量）对，我计划在分类过程中使用它们。

文件被加载到一个HashMap中，n-gram => weight。我们称该列表为 ngramWeights。我计划在 Spark 生成 n-gram 时在 ngramWeights 映射上进行查找。存储在地图中的 ngram 是 unigrams 和 bigrams。因此，我的 Spark 代码需要同时生成这两种情况。

高级方法：

• 使用 SQL 查询读取内容并“存储”在数据框中。
• 使用 MLLib 创建 Tokenizer 和 NGram 转换，N={1,2}
• 对于每一行
  • 以列表形式获取 ngram
  • 遍历列表并通过 ngramWeights 映射运行它，汇总得分
  • 返回一个 Tuple2 实例，以 row_id 作为键，总分作为值。

前 2 次运行的结果（N=1，N=2）存储在各自的 JavaPairRDD 结构中。

• 通过加入各自的JavaPairRDD，加入一元和二元分数。连接在 row_id 上。
  • 总分是两个权重之和。
• 在 JavaRDD 中返回结果

代码：

这是我用来为特定 N 的特定每一行评分的方法。

private static final JavaPairRDD<String, Double> scoreByNGrams( DataFrame dframe, int nGramSize, final Map<String, Double> ngramWeights ){

    Tokenizer tokenizer = new Tokenizer().setInputCol( "content" ).setOutputCol( "tokenized" );
    dframe = tokenizer.transform( dframe );

    NGram ngramTransform = new NGram().setN( nGramSize ).setInputCol( "tokenized" ).setOutputCol( "ngrams" );
    dframe = ngramTransform.transform( dframe );

    JavaPairRDD<String, Double> textScores = dframe.javaRDD().mapToPair(

        new PairFunction<Row, String, Double>(){

            public Tuple2<String, Double> call( Row row ) throws Exception{

                double score = 0.0;

                WrappedArray<String> wrappedArray = (WrappedArray<String>)row.getAs( "ngrams" );
                Iterator<String> iter = wrappedArray.iterator();
                while ( iter.hasNext() ){
                    String ngram = iter.next();
                    Double value = ngramWeights.get( ngram );
                    if ( value != null )
                        score += value;
                }

                String rowId = row.getAs( "row_id" ).toString();
                return new Tuple2<String, Double>( rowId, score );
            }
        });

    return textScores;
}

然后我按以下方式组合部分结果：

public static void main( String[] args ){

    Map<String, Double> ngramWeights = new HashMap<>();
    loadCoefficients( ngramWeights ); // simply reading from a file

    SparkConf sparkConf = new SparkConf().setAppName( "TestCase" );
    JavaSparkContext sc = new JavaSparkContext( sparkConf );
    HiveContext hiveContext = new HiveContext( sc.sc() );

    // Hive table is stored in PARQUET format
    hiveContext.setConf( "spark.sql.hive.convertMetastoreParquet", "false" );
    hiveContext.sql( "USE my_db" );

    DataFrame contentInfo = hiveContext.sql( "SELECT row_id, get_json_object( text, \"$.clean_text\" ) as content FROM mytable" );

    JavaPairRDD<String, Double> unigrams = scoreByNGrams( contentInfo, 1, ngramWeights );
    JavaPairRDD<String, Double> bigrams = scoreByNGrams( contentInfo, 2, ngramWeights );

    JavaPairRDD<String, Double> messageScores = unigrams.join( bigrams ).mapToPair( 

        new PairFunction<Tuple2<String, Tuple2<Double, Double>>(){

            @Override
            public Tuple2<String, Double> call( Tuple2<String, Tuple2<Double, Double>> t ){
                double sum = t._2()._1() + t._2()._2();
                return new Tuple2<String, Double>( t._1(), sum );
            }
        }
    );

    // I then proceed to store the result
    // ...
}

我编译我的代码，创建包含所有内部依赖项的胖 jar，将其放在 HDFS 上并使用以下命令运行它：

$ spark-submit --class my.package.MyClass --master yarn-cluster hdfs://my/hdfs/myfatjar.jar 

问题：

应用程序启动良好，似乎正在执行，但是当它达到 FINISHED 状态时，它立即抛出异常，说：

Exception in thread "main" org.apache.spark.SparkException: Application application_<id> finished with failed status
    at org.apache.spark.deploy.yarn.client.Client.run(Client.scala:1035)
    at org.apache.spark.deploy.yarn.client.Client$.main(Client.scala:1082)
    at org.apache.spark.deploy.yarn.client.Client.main(Client.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

仔细检查此应用程序的日志，发现失败的任务存在以下问题：

java.lang.NullPointerException
    at org.apache.spark.ml.feature.Tokenizer$$anonfun$createTransformFunc$1.apply(Tokenizer.scala:39)
    at org.apache.spark.ml.feature.Tokenizer$$anonfun$createTransformFunc$1.apply(Tokenizer.scala:39)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.evalExpr10$(Unknown Source)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
    at org.apache.spark.sql.execution.Project$$anonfun$1$$anonfun$apply$1.apply(basicOperators.scala:51)
    at org.apache.spark.sql.execution.Project$$anonfun$1$$anonfun$apply$1.apply(basicOperators.scala:49)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:194)
    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:64)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)

以前有人遇到过这个问题吗？我使用标记器的唯一地方是在我提供的函数中，所以我在那里看不到问题。

如果有人想知道，“text”字段是一个 json 字符串，总是有一个“clean_text”键。对 DB 运行以下查询：

SELECT * FROM mytable WHERE get_json_object( text, "$.clean_text" ) IS NULL

返回 0 个结果。因此，读取的内容并非为NULL。

如果有任何不清楚的地方或者您需要关于我正在执行的查询的更多信息，请告诉我。

---

更新：我删除了 ML 转换（tokenizer 和 ngram）并转换了代码，以便我可以使用自定义 NGram tokenizer（直接获取内容）。对于 一些 行，Spark 返回 NULL（作为字符串），尽管 在 mytable 中有 NO 个 null 值。

有人对此有任何想法吗？再次感谢！

Answer 1

我不确定这是否仍然会导致任何问题，但是我在使用 Databricks 时遇到了同样的错误，并发现问题是由 pyspark.ml.feature 模块中的 VectorAssembler 引起的。 在我的例子中，我为 inputCols 参数指定的列表与列出现在 VectorAssembler 正在转换的数据集中出现的顺序不同，这导致了问题。 通过确保我的 inputCols 列表的顺序与数据框列列表匹配，我停止了这个 java.lang.NullPointerException 错误的发生。