title: PySpark VectorAssembler 'Values to assemble cannot be null' error when there are no null values
body: <p><strong>TL;DR</strong>: Sometimes PySpark's <code>VectorAssembler</code> appears to throw an error in my Pipeline, and sometimes my Pipeline works fine. </p>
  
  <p>I'm using PySpark 2.1.0 (it's the only version supported by the IT team at work). My Pipeline consists of applying <code>StringIndexer</code> to categorical features, <code>OneHotEncoder</code> on these indexes, and <code>VectorAssembler</code> to combine these encodings with numerical features that don't need any pre-processing (they have already been normalized as part of my feature engineering -- I refer to them as "as-is cols" below). I then pass these features into a <code>DecisionTreeClassifier</code> algorithm, but my code faults out before this step, if it faults out. </p>
  
  <p>The error I receive, if I receive one, is: <code>Caused by: org.apache.spark.SparkException: Values to assemble cannot be null</code>. I have checked, and all the columns being used as features do not contain null values. (I impute them prior, and I have verified that this results in 0 nulls.)</p>
  
  <p>If the Pipeline works in a given kernel, it always works in that kernel. If the full Pipeline doesn't work in a given kernel, it always doesn't work in that kernel. If I break the non-working Pipeline into pieces -- i.e. process the categorical features separately from the numerical "as-is" ones -- sometimes the categorical pipeline works and the numerical one doesn't, and other times it's the opposite. </p>
  
  <p>I am loading data into PySpark from Hive using <code>sqlContext.sql("SELECT * FROM &lt;table&gt;")</code>. This <code>&lt;table&gt;</code> contains cleaned data &amp; engineered features and is not changing between successful and unsuccessful Pipelines. I am wondering if the order of the data differs upon each <code>SELECT *</code> call (i.e. every time I start a new kernel) and some data orders cause a fault -- though I haven't been able to discern a pattern. I found a <a href="https://www.matthewslipper.com/2017/10/01/watch-out-vectorassembler-incompatible-spark-streaming.html" rel="nofollow noreferrer">blog post</a> that incriminates <code>VectorAssembler</code> for inferring metadata from the first row of the dataset. But if my input contains no nulls, the order shouldn't matter, should it? The only thing I can think of is the first row containing an "empty" vector from <code>OneHotEncoder</code>'s <code>dropLast=True</code> behavior (e.g. <code>(2, [], [])</code>), but shouldn't <code>VectorAssembler</code> be able to handle this?</p>
  
  <p>Here is my code. I apologize in advance for not being able to provide data that recreates the error or results in a successful Pipeline, but my dataset is large &amp; proprietary. I can try to provide some snippets, if that would assist in trouble-shooting. Thanks.</p>
  
  <pre class="lang-py prettyprint-override"><code>from pyspark.ml.feature import OneHotEncoder, VectorAssembler, StringIndexer
  from pyspark.ml import Pipeline
  from pyspark.ml.classification import DecisionTreeClassifier
  
  # load data from Hive 
  train_df = sqlContext.sql('select * from &lt;hive table&gt;')
  
  # impute missing data
  asis_cols = [asis_col1, asis_col2, ...]
  ohe_cols = ['fac_usg', 'src_type']
  train_df_imp = (train_df.na.fill(0, subset=asis_cols)
                          .na.fill('blank', subset=ohe_cols))
  
  # create pipeline
  indexers = [StringIndexer(inputCol=c, outputCol=c+"_idxd") for c in ohe_cols]
  encoders = [OneHotEncoder(inputCol=c+"_idxd", outputCol=c+"_enc") for c in ohe_cols]
  
  assembly_list = asis_cols + [c+"_enc" for c in ohe_cols]
  assembler = VectorAssembler(inputCols=assembly_list, outputCol='features')
  
  labelIndexer = StringIndexer(inputCol='label', outputCol='indexedLabel', handleInvalid='skip')
  alg = DecisionTreeClassifier(labelCol='indexedLabel', minInstancesPerNode=3)
  
  pipe = Pipeline(stages=indexers + encoders + [assembler] + [labelIndexer] + [alg])
  
  # fit model
  model = pipe.fit(train_df_imp)
  </code></pre>
  
  <p><strong>Edit</strong>: Here is the schema of <code>train_df</code>. All <code>asis_col</code>'s are of type <code>double</code>.</p>
  
  <pre class="lang-py prettyprint-override"><code>root
   |-- rec_id: integer (nullable = true)
   |-- rec_date: date (nullable = true)
   |-- amount_usd: double (nullable = true)
   |-- src_type: string (nullable = true)
   |-- fac_usg: string (nullable = true)
   |-- description: string (nullable = true)
   |-- asis_col1: double (nullable = true)
   |-- asis_col2: double (nullable = true)
   ...
  </code></pre>
  
  <p>Example of the steps I took to confirm that <code>na.fill</code> worked:</p>
  
  <pre class="lang-py prettyprint-override"><code>from pyspark.sql.functions import count, isnan, when, col
  
  train_df_imp.select([count(when(isnan(c), c)).alias(c) for c in asis_cols]).toPandas().loc[0].sum()
  &gt;&gt;&gt; 0
  </code></pre>
  
  <p>Stack trace of error message:</p>
  
  <pre class="lang-py prettyprint-override"><code>Py4JJavaError: An error occurred while calling o1468.fit.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 86.0 failed 4 times, most recent failure: Lost task 0.3 in stage 86.0 (TID 4907, poldcdhdn010.dev.intranet, executor 4): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$3: (struct&lt;asis_col1:double,... 13 more fields&gt;) =&gt; vector)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:389)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
      at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
      at scala.collection.AbstractIterator.to(Iterator.scala:1336)
      at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
      at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336)
      at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
      at scala.collection.AbstractIterator.toArray(Iterator.scala:1336)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
      at org.apache.spark.scheduler.Task.run(Task.scala:99)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.spark.SparkException: Values to assemble cannot be null.
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160)
      at 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)
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:99)
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:98)
      ... 29 more
  
  Driver stacktrace:
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1435)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1423)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1422)
      at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
      at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
      at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1422)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)
      at scala.Option.foreach(Option.scala:257)
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:802)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1650)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1605)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1594)
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
      at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:628)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1918)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1931)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:1944)
      at org.apache.spark.rdd.RDD$$anonfun$take$1.apply(RDD.scala:1353)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
      at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
      at org.apache.spark.rdd.RDD.take(RDD.scala:1326)
      at org.apache.spark.ml.tree.impl.DecisionTreeMetadata$.buildMetadata(DecisionTreeMetadata.scala:112)
      at org.apache.spark.ml.tree.impl.RandomForest$.run(RandomForest.scala:105)
      at org.apache.spark.ml.classification.DecisionTreeClassifier.train(DecisionTreeClassifier.scala:116)
      at org.apache.spark.ml.classification.DecisionTreeClassifier.train(DecisionTreeClassifier.scala:45)
      at org.apache.spark.ml.Predictor.fit(Predictor.scala:96)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
      at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
      at py4j.Gateway.invoke(Gateway.java:280)
      at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
      at py4j.commands.CallCommand.execute(CallCommand.java:79)
      at py4j.GatewayConnection.run(GatewayConnection.java:214)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$3: (struct&lt;asis_col1:double,... 13 more fields&gt;) =&gt; vector)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:377)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:389)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
      at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
      at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
      at scala.collection.AbstractIterator.to(Iterator.scala:1336)
      at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
      at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336)
      at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
      at scala.collection.AbstractIterator.toArray(Iterator.scala:1336)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
      at org.apache.spark.rdd.RDD$$anonfun$take$1$$anonfun$29.apply(RDD.scala:1353)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
      at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1944)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
      at org.apache.spark.scheduler.Task.run(Task.scala:99)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      ... 1 more
  Caused by: org.apache.spark.SparkException: Values to assemble cannot be null.
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$assemble$1.apply(VectorAssembler.scala:160)
      at 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)
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:99)
      at org.apache.spark.ml.feature.VectorAssembler$$anonfun$3.apply(VectorAssembler.scala:98)
      ... 29 more
  </code></pre>