Spark VectorAssembler 错误 - PySpark 2.3 - Python

Question

我正在使用 pySpark 2.3.0，并创建了一个非常简单的 Spark 数据框来测试 VectorAssembler 的功能。这是一个更大的数据框的子集，我只选择了几个数字（双数据类型）列：

>>>cols = ['index','host_listings_count','neighbourhood_group_cleansed',\
        'bathrooms','bedrooms','beds','square_feet', 'guests_included',\
        'review_scores_rating']
>>>test = df[cols]
>>>test.take(3)

[行（索引=0，host_listings_count=1，neighborhood_group_cleansed=无，浴室=1.5，卧室=2.0，床位=3.0，square_feet=无，guests_included=1，review_scores_rating=100.0），行（索引=1，host_listings_count=1， neighborhood_group_cleansed=无，浴室=1.5，卧室=2.0，床位=3.0，square_feet=无，guests_included=1，review_scores_rating=100.0），行（索引=2，host_listings_count=1，neighborhood_group_cleansed=None.5，卧室=1 ，床位=3.0，square_feet=无，guests_included=1，review_scores_rating=100.0)]

从上面看来，这个 Spark 数据框没有任何问题。所以我然后创建如下所示的汇编程序并得到显示的错误。可能出了什么问题？

>>>from pyspark.ml.feature import VectorAssembler
>>>assembler = VectorAssembler(inputCols=cols, outputCol="features")
>>>output = assembler.transform(test)
>>>output.take(3)

Py4JJavaError：调用 o279.collectToPython 时出错。：org.apache.spark.SparkException：作业因阶段失败而中止：阶段 5.0 中的任务 0 失败 1 次，最近失败：阶段 5.0 中丢失任务 0.0（TID 10、本地主机、执行程序驱动程序）：org.apache.spark .SparkException: 无法在 org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source) at org.apache.spark 执行用户定义的函数（$anonfun$3: (struct) => vector）。 sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) 在 org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377) 在 org.apache.spark。 sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:231) 在 org.apache.spark.sql.execution.SparkPlan$$anonfun$2。Thread.run(Thread.java:748) 导致：org.apache.spark.SparkException：要组装的值不能为空。在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler. scala:143) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) at org.apache.spark.ml.feature.VectorAssembler $.assemble(VectorAssembler.scala:143) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:99) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun $3.apply(VectorAssembler.scala:98) ... 16 更多 要组装的值不能为空。在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler. scala:143) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) at org.apache.spark.ml.feature.VectorAssembler $.assemble(VectorAssembler.scala:143) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:99) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun $3.apply(VectorAssembler.scala:98) ... 16 更多 要组装的值不能为空。在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler. scala:143) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) at org.apache.spark.ml.feature.VectorAssembler $.assemble(VectorAssembler.scala:143) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:99) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun $3.apply(VectorAssembler.scala:98) ... 16 更多

Executor$TaskRunner.run(Executor.scala:322) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) .. . 1 more 引起：org.apache.spark.SparkException：要组装的值不能为空。在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler. scala:143) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35) at org.apache.spark.ml.feature.VectorAssembler $.assemble(VectorAssembler.scala:143) 在 org.apache.spark.ml.feature.VectorAssembler$$anonfun$3。

Answer 1

您发布的堆栈跟踪提到问题是由正在组装的列中的空值引起的。

您需要处理列中的null值cols。test.fillna(0, subset=cols)在调用转换之前尝试，或者过滤掉那些列中具有空值的行。