title: How can use sqlContext in spark udf
body: <p>I'm trying to load a json file in spark UDF and use it to query something. What i need to do is to do use a column value (storeId) from a data frame and  use it in </p>
  
  <p>But, i get a pickle error. If I try coding without the sqlContext then it work.</p>
  
  <p>Is there any workaround or is this not possible. </p>
  
  <pre><code>def get_id_udf (storeId,sqlContext):
  
     df = sqlContext.read.json("file_url_s3")
  
     if storeId == None:
         return None
  
     return None
  
  from pyspark.sql.functions import udf, col
  
  from pyspark.sql.types import IntegerType
  
  desc_udf = udf(lambda storeId : get_id_udf(storeId,sqlContext),       IntegerType())
  </code></pre>
  
  <blockquote>
  <pre><code>---------------------------------------------------------------------------
  TypeError                                 Traceback (most recent call last)
  &lt;ipython-input-22-b5c4070c110e&gt; in &lt;module&gt;()
        1 from pyspark.sql.functions import udf, col
        2 from pyspark.sql.types import IntegerType
  ----&gt; 3 desc_udf = udf(lambda storeId : get_cluster_id_udf(storeId,sqlContext), IntegerType())
  
  /usr/lib/spark/python/pyspark/sql/functions.py in udf(f, returnType)
     1799     [Row(slen=5), Row(slen=3)]
     1800     """
  -&gt; 1801     return UserDefinedFunction(f, returnType)
     1802 
     1803 blacklist = ['map', 'since', 'ignore_unicode_prefix']
  
  /usr/lib/spark/python/pyspark/sql/functions.py in __init__(self, func, returnType, name)
     1758         self.returnType = returnType
     1759         self._broadcast = None
  -&gt; 1760         self._judf = self._create_judf(name)
     1761 
     1762     def _create_judf(self, name):
  
  /usr/lib/spark/python/pyspark/sql/functions.py in _create_judf(self, name)
     1763         from pyspark.sql import SQLContext
     1764         sc = SparkContext.getOrCreate()
  -&gt; 1765         wrapped_func = _wrap_function(sc, self.func, self.returnType)
     1766         ctx = SQLContext.getOrCreate(sc)
     1767         jdt = ctx._ssql_ctx.parseDataType(self.returnType.json())
  
  /usr/lib/spark/python/pyspark/sql/functions.py in _wrap_function(sc, func, returnType)
     1743 def _wrap_function(sc, func, returnType):
     1744     command = (func, returnType)
  -&gt; 1745     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)
     1746     return sc._jvm.PythonFunction(bytearray(pickled_command), env, includes,
  </code></pre>
    
    <p>sc.pythonExec,
           1747                                   sc.pythonVer, broadcast_vars, sc._javaAccumulator)</p>
  
  <pre><code>/usr/lib/spark/python/pyspark/rdd.py in _prepare_for_python_RDD(sc, command)
     2313     # the serialized command will be compressed by broadcast
     2314     ser = CloudPickleSerializer()
  -&gt; 2315     pickled_command = ser.dumps(command)
     2316     if len(pickled_command) &gt; (1 &lt;&lt; 20):  # 1M
     2317         # The broadcast will have same life cycle as created PythonRDD
  
  /usr/lib/spark/python/pyspark/serializers.py in dumps(self, obj)
      426 
      427     def dumps(self, obj):
  --&gt; 428         return cloudpickle.dumps(obj, 2)
      429 
      430 
  
  /usr/lib/spark/python/pyspark/cloudpickle.py in dumps(obj, protocol)
      655 
      656     cp = CloudPickler(file,protocol)
  --&gt; 657     cp.dump(obj)
      658 
      659     return file.getvalue()
  
  /usr/lib/spark/python/pyspark/cloudpickle.py in dump(self, obj)
      105         self.inject_addons()
      106         try:
  --&gt; 107             return Pickler.dump(self, obj)
      108         except RuntimeError as e:
      109             if 'recursion' in e.args[0]:
  
  /usr/lib64/python2.7/pickle.pyc in dump(self, obj)
      222         if self.proto &gt;= 2:
      223             self.write(PROTO + chr(self.proto))
  --&gt; 224         self.save(obj)
      225         self.write(STOP)
      226 
  
  
  /usr/lib64/python2.7/pickle.pyc in save(self, obj)
      284         f = self.dispatch.get(t)
      285         if f:
  --&gt; 286             f(self, obj) # Call unbound method with explicit self
      287             return
      288 
  
  /usr/lib64/python2.7/pickle.pyc in save_dict(self, obj)
      653 
      654         self.memoize(obj)
  --&gt; 655         self._batch_setitems(obj.iteritems())
      656 
      657     dispatch[DictionaryType] = save_dict
  
  /usr/lib64/python2.7/pickle.pyc in _batch_setitems(self, items)
      685                 for k, v in tmp:
      686                     save(k)
  --&gt; 687                     save(v)
      688                 write(SETITEMS)
      689             elif n:
  
  /usr/lib64/python2.7/pickle.pyc in save(self, obj)
      304             reduce = getattr(obj, "__reduce_ex__", None)
      305             if reduce:
  --&gt; 306                 rv = reduce(self.proto)
      307             else:
      308                 reduce = getattr(obj, "__reduce__", None)
  
  TypeError: 'JavaPackage' object is not callable
  </code></pre>
  </blockquote>