title: Pyspark rdd.toDF() Data Loss
body: <p>I have an RDD created with a fairly complex mapping function that takes advantage of a large, broadcast dictionary.  If I look at <code>rdd.filter(lambda x: x... == CONDITION).collect()</code> to extract a couple cases I've seen that have been representative of my data loss issue, the data is correct and complete.  When I then try <code>df = rdd.toDF()</code>, and then <code>df.where(CONDITION).collect()</code>, large portions of many elements of the data have been replaced by <code>None</code>.</p>
  
  <p>If this gives any hint, the rows/elements are feature sets, and each row is created in blocks, with separate calls to the broadcasted dictionary for each block.  When the data is cast from a number (sometimes all reals, sometimes all zeros, sometimes 1-hot) to None, it seems to always be done over one/some of the complete blocks in which the row/element vector is patched together.</p>
  
  <p>To give a better example of the code</p>
  
  <pre><code>def map_fxn(row, n_dim, dict_bc, row_constr):
    if row['key'] in dict_bc.value:
      rvec = dict_bc.value[row['key']
    else:
      rvec = [0] * n_dim
    return row_constr(row['id'], *rvec)
  
  my_dict = {'key1' : range(10,20), 'key2' : range(20,30)}
  n_dim = 10
  dict_bc = spark.sparkContext.broadcast(my_dict)
  rdd = spark.sparkContext.parallelize([Row(key='key1', id='id1'), 
                                        Row(key='key3', id='id3')])
  row_constr = Row('id', *['val' + str(i) for i in range(n_dim)])
  rdd_map = rdd.map(lambda x: map_fxn(x, n_dim, dict_bc, row_constr)
  df = rdd_map.toDF()
  </code></pre>
  
  <p>Then <code>rdd_map</code> is fine.  <code>df</code> has random seeming rows with <code>None</code> values instead of either 0s or the values from <code>dict_bc.value</code>.  The dictionary is loaded from S3 output of mllib, where every entry is guaranteed to be a double.  So there are no strings anywhere.  I've tried casting the RDD to a DF using the <code>createDataFrame</code> function both from <code>SparkSession</code> and from <code>SQLContext</code>, with a schema built manually.  Both return fine but then error out with unhelpful, failed n times, error messages if you try to actually do anything at all with the dataframe.</p>