title: Py4J can't serialize PySpark UDF
body: <p>I'm trying to calculate the Euclidean distance from a random vector for every row in a spark dataframe. Initially a pandas df called omatrix, each column is a word, and each row is a sentence represented as a vector of 1's and 0's, where 1 means the word is present and 0 otherwise.</p>
  
  <pre><code>import numpy as np
  
  randv = np.random.rand(len(omatrix.columns))
  
  def distance(x):
      return np.linalg.norm(x-randv).item()
  
  dist = udf(distance, FloatType())
  
  df = spark.createDataFrame([[word] for word in omatrix.values.tolist()], schema=["features"])
  df.printSchema()
  df.show(5)
  </code></pre>
  
  <p>This successfully outputs:</p>
  
  <pre><code>root
   |-- features: array (nullable = true)
   |    |-- element: long (containsNull = true)
  
  +--------------------+
  |            features|
  +--------------------+
  |[0, 0, 0, 0, 0, 0...|
  |[0, 0, 0, 0, 0, 0...|
  |[0, 1, 0, 0, 1, 0...|
  |[0, 0, 0, 0, 0, 1...|
  |[0, 1, 0, 1, 0, 0...|
  +--------------------+
  only showing top 5 rows
  </code></pre>
  
  <p>Now I thought I could use withColumn() to calculate the distances:</p>
  
  <pre><code>df = df.withColumn('distance', dist(df.features))
  df.select(df.distance).show(5)
  </code></pre>
  
  <p>The second line results in the stacktrace:</p>
  
  <pre><code>---------------------------------------------------------------------------
  Py4JJavaError                             Traceback (most recent call last)
  &lt;ipython-input-27-1a55784bd46e&gt; in &lt;module&gt;()
  ----&gt; 1 vdf.select(vdf.distance).show(5)
  
  /home/jakeu123/spark/python/pyspark/sql/dataframe.pyc in show(self, n, truncate, vertical)
      348         """
      349         if isinstance(truncate, bool) and truncate:
  --&gt; 350             print(self._jdf.showString(n, 20, vertical))
      351         else:
      352             print(self._jdf.showString(n, int(truncate), vertical))
  
  /home/jakeu123/spark/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py in __call__(self, *args)
     1158         answer = self.gateway_client.send_command(command)
     1159         return_value = get_return_value(
  -&gt; 1160             answer, self.gateway_client, self.target_id, self.name)
     1161 
     1162         for temp_arg in temp_args:
  
  /home/jakeu123/spark/python/pyspark/sql/utils.pyc in deco(*a, **kw)
       61     def deco(*a, **kw):
       62         try:
  ---&gt; 63             return f(*a, **kw)
       64         except py4j.protocol.Py4JJavaError as e:
       65             s = e.java_exception.toString()
  
  /home/jakeu123/spark/python/lib/py4j-0.10.6-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
      318                 raise Py4JJavaError(
      319                     "An error occurred while calling {0}{1}{2}.\n".
  --&gt; 320                     format(target_id, ".", name), value)
      321             else:
      322                 raise Py4JError(
  
  Py4JJavaError: An error occurred while calling o311.showString.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 11.0 failed 4 times, most recent failure: Lost task 0.3 in stage 11.0 (TID 24, 10.142.0.5, executor 0): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 218, in main
      func, profiler, deserializer, serializer = read_udfs(pickleSer, infile, eval_type)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 138, in read_udfs
      arg_offsets, udf = read_single_udf(pickleSer, infile, eval_type)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 118, in read_single_udf
      f, return_type = read_command(pickleSer, infile)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 58, in read_command
      command = serializer._read_with_length(file)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 170, in _read_with_length
      return self.loads(obj)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 562, in loads
      return pickle.loads(obj)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 929, in subimport
      __import__(name)
  ImportError: ('No module named numpy.linalg.linalg', &lt;function subimport at 0x7ff875f3c6e0&gt;, ('numpy.linalg.linalg',))
  
      at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
      at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:83)
      at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:66)
      at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
      at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
      at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:253)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
      at org.apache.spark.scheduler.Task.run(Task.scala:109)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
      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)
  
  Driver stacktrace:
      at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1599)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1587)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1586)
      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:1586)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
      at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:831)
      at scala.Option.foreach(Option.scala:257)
      at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:831)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1820)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1769)
      at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1758)
      at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
      at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:642)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:2027)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:2048)
      at org.apache.spark.SparkContext.runJob(SparkContext.scala:2067)
      at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:363)
      at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
      at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3272)
      at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
      at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2484)
      at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3253)
      at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:77)
      at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3252)
      at org.apache.spark.sql.Dataset.head(Dataset.scala:2484)
      at org.apache.spark.sql.Dataset.take(Dataset.scala:2698)
      at org.apache.spark.sql.Dataset.showString(Dataset.scala:254)
      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:282)
      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.api.python.PythonException: Traceback (most recent call last):
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 218, in main
      func, profiler, deserializer, serializer = read_udfs(pickleSer, infile, eval_type)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 138, in read_udfs
      arg_offsets, udf = read_single_udf(pickleSer, infile, eval_type)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 118, in read_single_udf
      f, return_type = read_command(pickleSer, infile)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/worker.py", line 58, in read_command
      command = serializer._read_with_length(file)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 170, in _read_with_length
      return self.loads(obj)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 562, in loads
      return pickle.loads(obj)
    File "/home/jakeu123/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 929, in subimport
      __import__(name)
  ImportError: ('No module named numpy.linalg.linalg', &lt;function subimport at 0x7ff875f3c6e0&gt;, ('numpy.linalg.linalg',))
  
      at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
      at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:83)
      at org.apache.spark.sql.execution.python.PythonUDFRunner$$anon$1.read(PythonUDFRunner.scala:66)
      at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
      at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
      at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
      at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
      at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
      at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:253)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:830)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
      at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
      at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
      at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
      at org.apache.spark.scheduler.Task.run(Task.scala:109)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      ... 1 more
  </code></pre>
  
  <p>Why is Py4J trying to import numpy.linalg.linalg? Or is it another issue that is causing this?</p>