Apache Spark lancia NullPointerException quando si incontra la funzione mancante

Question

Ho un bizzarro problema con PySpark durante l'indicizzazione della colonna di stringhe nelle funzionalità. Ecco il mio file 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 

dove ho un valore mancante per 'x0'. In un primo momento, sto leggendo le caratteristiche da file CSV in dataframe utilizzando pyspark_csv: https://github.com/seahboonsiew/pyspark-csv quindi indicizzare x0 con StringIndexer:

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() 

quando si chiama '' ind.collect() '' Spark getta java.lang.NullPointerException . Tutto funziona bene per set di dati completi, ad esempio, per 'x1' però.

Qualcuno ha idea di cosa sta causando questo e come risolverlo?

Grazie in anticipo!

Sergey

Aggiornamento:

Io uso Spark 1.5.1. L'errore esatto:

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$2.apply(Metadata.scala:196) 
at org.apache.spark.sql.types.Metadata$$anonfun$org$apache$spark$sql$types$Metadata$$hash$2.apply(Metadata.scala:196) 
... etc 

ho cercato di creare la stessa dataframe senza file CSV lettura,

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']) 

e dà lo stesso errore. Un esempio po 'diverso funziona bene,

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| 
+---+----------+----+------+ 

Aggiornamento 2:

Ho appena scoperto lo stesso problema a Scala, quindi credo che sia bug Spark non solo PySpark. In particolare, frame di dati

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") 

tiri java.lang.NullPointerException durante l'indicizzazione 'x0' caratteristica. Inoltre, quando l'indicizzazione 'x0' nel seguente frame di dati

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") 

c'ho 'java.lang.UnsupportedOperationException: Schema di tipo Qualsiasi Non è supportato' che è causato dalla mancanza di valore 'num' in 5th vettoriale . Se lo si sostituisce con un numero, tutto funziona bene anche se manca il valore nel 1 ° vettore.

Ho anche provato versioni precedenti di Spark (1.4.1), e il risultato è lo stesso.

Answer 1

Sembra che il modulo che si sta utilizzando converta stringhe vuote in valori null e che a un certo punto produca problemi con l'elaborazione downstream. A prima vista it looks like a PySpark bug.

Come ripararlo? Una soluzione semplice è quella di entrambi i null goccia prima di indicizzazione:

features.na.drop() 

o sostituire i null con alcuni segnaposto:

from pyspark.sql.functions import col, when 

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

Inoltre, è possibile utilizzare spark-csv.È efficiente, testato e come bonus non converte stringhe vuote a nulls.

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

Answer 2

Beh, attualmente, l'unica soluzione è quella di sbarazzarsi di NA di come @ zero323 proposto o per convertire Spark dataframe per Pandas dataframe utilizzando toPandas() e imputare i dati utilizzando sklearn Imputer o qualsiasi imputer personalizzato, ad esempio, , Impute categorical missing values in scikit-learn, quindi convertire Pandas Dataframe in Spark DataFrame e lavorare con esso. Tuttavia, il problema rimane, proverò a inviare una segnalazione di bug, se presente. Sono relativamente nuovo a Spark, quindi c'è una possibilità che mi manchi qualcosa.