title: User Defined Function breaks pyspark dataframe
body: <p>My spark version is 1.3, I am using pyspark.</p>
  
  <p>I have a large dataframe called df.</p>
  
  <pre><code>from pyspark import SQLContext
  sqlContext = SQLContext(sc)
  df = sqlContext.parquetFile("events.parquet")
  </code></pre>
  
  <p>I then select a few columns of the dataframe and try to count the number of rows.  This works fine.</p>
  
  <pre><code>df3 = df.select("start", "end", "mrt")
  print(type(df3))
  print(df3.count())
  </code></pre>
  
  <p>I then apply a user defined function to convert one of the columns from a string to a number, this also works fine</p>
  
  <pre><code>from pyspark.sql.functions import UserDefinedFunction
  from pyspark.sql.types import LongType
  CtI = UserDefinedFunction(lambda i: int(i), LongType())
  df4 = df2.withColumn("mrt-2", CtI(df2.mrt))
  </code></pre>
  
  <p>However if I try to count the number of rows I get an exception even though the type shows that it is a dataframe just like df3.</p>
  
  <pre><code>print(type(df4))
  print(df4.count())
  </code></pre>
  
  <p>My Error:</p>
  
  <pre><code>---------------------------------------------------------------------------
  Py4JJavaError                             Traceback (most recent call last)
  &lt;ipython-input-10-53941e183807&gt; in &lt;module&gt;()
        8 df4 = df2.withColumn("mrt-2", CtI(df2.mrt))
        9 print(type(df4))
  ---&gt; 10 print(df4.count())
       11 df3 = df4.select("start", "end", "mrt-2").withColumnRenamed("mrt-2", "mrt")
  
  /data/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/lib/spark/python/pyspark/sql/dataframe.py in count(self)
      299         2L
      300         """
  --&gt; 301         return self._jdf.count()
      302 
      303     def collect(self):
  
  /data/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in __call__(self, *args)
      536         answer = self.gateway_client.send_command(command)
      537         return_value = get_return_value(answer, self.gateway_client,
  --&gt; 538                 self.target_id, self.name)
      539 
      540         for temp_arg in temp_args:
  
  /data/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
      298                 raise Py4JJavaError(
      299                     'An error occurred while calling {0}{1}{2}.\n'.
  --&gt; 300                     format(target_id, '.', name), value)
      301             else:
      302                 raise Py4JError(
  
  Py4JJavaError: An error occurred while calling o152.count.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 1379 in stage 12.0 failed 4 times, most recent failure: Lost task 1379.3 in stage 12.0 (TID 27021, va1ccogbds01.lab.ctllabs.io): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "/data/0/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/jars/spark-assembly-1.3.0-cdh5.4.7-hadoop2.6.0-cdh5.4.7.jar/pyspark/worker.py", line 101, in main
      process()
    File "/data/0/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/jars/spark-assembly-1.3.0-cdh5.4.7-hadoop2.6.0-cdh5.4.7.jar/pyspark/worker.py", line 96, in process
      serializer.dump_stream(func(split_index, iterator), outfile)
    File "/data/0/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/jars/spark-assembly-1.3.0-cdh5.4.7-hadoop2.6.0-cdh5.4.7.jar/pyspark/serializers.py", line 236, in dump_stream
      vs = list(itertools.islice(iterator, batch))
    File "/data/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/lib/spark/python/pyspark/sql/functions.py", line 119, in &lt;lambda&gt;
    File "&lt;ipython-input-10-53941e183807&gt;", line 7, in &lt;lambda&gt;
  TypeError: int() argument must be a string or a number, not 'NoneType'
  
  at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:135)
  at org.apache.spark.api.python.PythonRDD$$anon$1.next(PythonRDD.scala:98)
  at org.apache.spark.api.python.PythonRDD$$anon$1.next(PythonRDD.scala:94)
  at org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:43)
  at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
  at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
  at org.apache.spark.rdd.RDD$$anonfun$zip$1$$anon$1.hasNext(RDD.scala:743)
  at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
  at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
  at org.apache.spark.sql.execution.Aggregate$$anonfun$execute$1$$anonfun$6.apply(Aggregate.scala:127)
  at org.apache.spark.sql.execution.Aggregate$$anonfun$execute$1$$anonfun$6.apply(Aggregate.scala:124)
  at org.apache.spark.rdd.RDD$$anonfun$14.apply(RDD.scala:634)
  at org.apache.spark.rdd.RDD$$anonfun$14.apply(RDD.scala:634)
  at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
  at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:277)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:244)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
  at org.apache.spark.scheduler.Task.run(Task.scala:64)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)
  
  Driver stacktrace:
  at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1210)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1199)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1198)
  at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
  at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
  at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1198)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
  at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:693)
  at scala.Option.foreach(Option.scala:236)
  at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:693)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1400)
  at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1361)
  at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
  ---------------------------------------------------------------------------
  </code></pre>
  
  <p>Am I using the user defined function correctly?  Any idea why the data frame functions don't work on the data frame?</p>