title: Pyspark - Calling function within lambda causes import error
body: <p>I'm trying to do something fairly simple. I've got a datetime object as part of my dataframe, and when doing a map, I'd like to format the date a specific way. I've created a custom function:</p>
  
  <pre><code>def format_date(dt):
      """Set this for date formatting. dt is datetime."""
      return dt.strftime("%Y/%m/%d %H:%M:%S")
  </code></pre>
  
  <p>And then later on, I use this in my map call (x.t is a datetime object):</p>
  
  <pre><code>unique = df.map(lambda x: (x.id,[[format_date(x.t),x.val]]))\
        .reduceByKey(lambda x,y: x+y)\
        .collectAsMap()
  </code></pre>
  
  <p>This causes the following exception <em>when submitted as a job</em>:</p>
  
  <pre><code>An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
  : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 4 times, most recent failure: Lost task 0.3 in stage 2.0 (TID 9, preteckt1.softlayer.com): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "/opt/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
      command = pickleSer._read_with_length(infile)
    File "/opt/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
      return self.loads(obj)
    File "/opt/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
      return pickle.loads(obj)
  ImportError: No module named analysis
  </code></pre>
  
  <p>Note that my script name is "run_analyses.py", and imports all of the functions from "analysis.py". I submit the job with </p>
  
  <pre><code>/opt/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --total-executor-cores 12 run_analyses.py
  </code></pre>
  
  <p>The strangest thing is that it works perfectly fine if I copy the code to an interactive pyspark session (or if I remove the format_date call). I can get around this by creating a new column and using UDF on my format_date function to create a new column, but I'd like to know why this approach is failing.</p>
  
  <p>I've pasted the more complete code below.</p>
  
  <p><strong>Edit</strong>: It appears it succeeds if I run the code directly from analysis.py, but fails if I run it from run_analysis.py. I've altered the code below to more accurately show this.</p>
  
  <p>run_analyses.py</p>
  
  <pre><code>import datetime, json, math, subprocess
  from os.path import expanduser
  from pyspark import SparkContext
  from pyspark.sql import SQLContext, HiveContext
  from analysis import *
  
  sc = SparkContext()
  sqlCtx = HiveContext(sc)
  ids = {}
  ...
  my_func(sqlCtx,ids)
  </code></pre>
  
  <p>analysis.py</p>
  
  <pre><code>def my_func(sqlCtx,ids):
      df = sqlCtx.read.format("org.apache.spark.sql.cassandra").load(table="table_name", keyspace="keyspace_name").select("id","t","val")
      df = df.filter((df.t &gt; last_week)&amp;(df.t &lt; now))
      df = df.filter(df.val &gt; 0)
      write_vals(df)
      ...
  
  def write_vals(df):
      unique = df.map(lambda x: (x.id,[[format_date(x.t),x.val]]))\
              .reduceByKey(lambda x,y: x+y)\
              .collectAsMap()
      ...
      return
  </code></pre>