Apache Spark 在遇到缺少的功能时抛出 NullPointerException

Question

在对功能中的字符串列进行索引时，PySpark 出现了一个奇怪的问题。这是我的 tmp.csv 文件：

x0,x1,x2,x3 
asd2s,1e1e,1.1,0
asd2s,1e1e,0.1,0
,1e3e,1.2,0
bd34t,1e1e,5.1,1
asd2s,1e3e,0.2,0
bd34t,1e2e,4.3,1

其中 'x0' 有一个缺失值。首先，我使用 pyspark_csv: https://github.com/seahboonsiew/pyspark-csv 将 csv 文件中的特征读入 DataFrame 然后用 StringIndexer:

索引 x0

import pyspark_csv as pycsv
from pyspark.ml.feature import StringIndexer

sc.addPyFile('pyspark_csv.py')

features = pycsv.csvToDataFrame(sqlCtx, sc.textFile('tmp.csv'))
indexer = StringIndexer(inputCol='x0', outputCol='x0_idx' )
ind = indexer.fit(features).transform(features)
print ind.collect()

调用“'ind.collect()'”时，Spark 抛出 java.lang.NullPointerException。对于完整的数据集，一切都很好，例如，'x1'。

有谁知道是什么原因造成的以及如何解决？

提前致谢！

谢尔盖

更新：

我使用 Spark 1.5.1。确切的错误：

File "/spark/spark-1.4.1-bin-hadoop2.6/python/pyspark/sql/dataframe.py", line 258, in show
print(self._jdf.showString(n))

File "/spark/spark-1.4.1-bin-hadoop2.6/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 538, in __call__

File "/spark/spark-1.4.1-bin-hadoop2.6/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py", line 300, in get_return_value

py4j.protocol.Py4JJavaError: An error occurred while calling o444.showString.
: java.lang.NullPointerException
at org.apache.spark.sql.types.Metadata$.org$apache$spark$sql$types$Metadata$$hash(Metadata.scala:208)
at org.apache.spark.sql.types.Metadata$$anonfun$org$apache$spark$sql$types$Metadata$$hash.apply(Metadata.scala:196)
at org.apache.spark.sql.types.Metadata$$anonfun$org$apache$spark$sql$types$Metadata$$hash.apply(Metadata.scala:196)
... etc

我试图在不读取 csv 文件的情况下创建相同的 DataFrame，

df = sqlContext.createDataFrame(
  [('asd2s','1e1e',1.1,0), ('asd2s','1e1e',0.1,0), 
  (None,'1e3e',1.2,0), ('bd34t','1e1e',5.1,1), 
  ('asd2s','1e3e',0.2,0), ('bd34t','1e2e',4.3,1)],
  ['x0','x1','x2','x3'])

它给出了同样的错误。有点不同的例子效果很好，

df = sqlContext.createDataFrame(
  [(0, None, 1.2), (1, '06330986ed', 2.3), 
  (2, 'b7584c2d52', 2.5), (3, None, .8), 
  (4, 'bd17e19b3a', None), (5, '51b5c0f2af', 0.1)],
  ['id', 'x0', 'num'])

// after indexing x0

+---+----------+----+------+
| id|        x0| num|x0_idx|
+---+----------+----+------+
|  0|      null| 1.2|   0.0|
|  1|06330986ed| 2.3|   2.0|
|  2|b7584c2d52| 2.5|   4.0|
|  3|      null| 0.8|   0.0|
|  4|bd17e19b3a|null|   1.0|
|  5|51b5c0f2af| 0.1|   3.0|
+---+----------+----+------+

更新二：

我刚刚在 Scala 中发现了同样的问题，所以我猜这是 Spark 错误，而不仅仅是 PySpark。特别是数据框

val df = sqlContext.createDataFrame(
  Seq(("asd2s","1e1e",1.1,0), ("asd2s","1e1e",0.1,0), 
      (null,"1e3e",1.2,0), ("bd34t","1e1e",5.1,1), 
      ("asd2s","1e3e",0.2,0), ("bd34t","1e2e",4.3,1))
).toDF("x0","x1","x2","x3")

索引 'x0' 特征时抛出 java.lang.NullPointerException。此外，在以下数据帧

中索引'x0'时

val df = sqlContext.createDataFrame(
  Seq((0, null, 1.2), (1, "b", 2.3), 
      (2, "c", 2.5), (3, "a", 0.8), 
      (4, "a", null), (5, "c", 0.1))
).toDF("id", "x0", "num")

我得到了 'java.lang.UnsupportedOperationException: Schema for type Any is not supported'，这是由于第 5 个向量中缺少 'num' 值造成的。如果用一个数字替换它，即使在第一个向量中缺少值，一切都很好。

我也试过旧版本的 Spark (1.4.1)，结果是一样的。

Answer 1

看起来您正在使用的模块将空字符串转换为 null，并且它在某些时候会干扰下游处理。乍一看it looks like a PySpark bug.

如何解决？一个简单的解决方法是在索引之前删除空值：

features.na.drop()

或用一些占位符替换空值：

from pyspark.sql.functions import col, when

features.withColumn(
    "x0", when(col("x0").isNull(), "__SOME_PLACEHOLDER__").otherwise(col("x0")))

此外，您可以使用 spark-csv。它是高效的，经过测试并且作为奖励不会将空字符串转换为 nulls.

features = (sqlContext.read
    .format('com.databricks.spark.csv')
    .option("inferSchema", "true")
    .option("header", "true")
    .load("tmp.csv"))

Answer 2

好吧，目前唯一的解决方案是摆脱@zero323 提议的 NA 或使用 toPandas() 方法将 Spark DataFrame 转换为 Pandas DataFrame 并使用 sklearn 估算数据输入器或任何自定义输入器，例如 Impute categorical missing values in scikit-learn，然后将 Pandas Dataframe 转换回 Spark DataFrame 并使用它。尽管如此，问题仍然存在，如果有的话，我会尝试提交错误报告。我对 Spark 比较陌生，所以我有可能遗漏了一些东西。

Apache Spark 在遇到缺少的功能时抛出 NullPointerException

Apache Spark throws NullPointerException when encountering missing feature

python

apache-spark

apache-spark-sql

pyspark

apache-spark-ml