title: Pyspark collectAsMap() UDAF alternative - Could not serialize object: Py4JError: An error occurred while calling o62.__getstate__
body: <p>I am trying to apply aggregated function to columns in pyspark. The context is I have only Spark 2.2 at hand and no option of using vectorized pandas_udf</p>
  
  <pre><code>sdt = spark.createDataFrame(zip([random.randint(1,100) for x in range(20)], [random.randint(1,100) for x in range(20)]), schema=['col1', 'col2'])
  +----+----+
  |col1|col2|
  +----+----+
  |  19|  51|
  |  95|  56|
  |  11|  94|
  |  80|  99|
  |  20|  80|
  |  38|  91|
  |  18|  88|
  |   4|  33|
  +----+----+
  </code></pre>
  
  <p>In order to parallelize the columns, I convert it into rdd</p>
  
  <pre><code>sdt_col_rdd = sc.parallelize(sdt.columns)
  </code></pre>
  
  <p>Testing works fine with ordinary python functions, returning panda's dataframe</p>
  
  <pre><code>x = sdt_col_rdd.map(lambda col : (col, pd.DataFrame(np.random.randint(0,100,size=(2, 4)), columns=list('ABCD'))))
  y = x.collectAsMap() #collect into dictionary with column names as key
  print(y['col1']; print(y['col2']);
      A   B   C   D
  0  14  55   4  57
  1  36  84  53  51
      A   B   C   D
  0  14  55   4  57
  1  36  84  53  51
  </code></pre>
  
  <p>Switching into spark dataframe, here's also a sample function returning panda's df, but process Spark's df and use it's native aggregations, transformation, actions, etc. :</p>
  
  <pre><code>def myFunc(df, c):
      #other more processing, aggregation, transformation may be performed here
      res = df.agg((F.min(c)-1).alias("min_"+c), (F.max(c)+1).alias("max_"+c)).toPandas()
      res["col_name"] = c
      return res
  </code></pre>
  
  <p>Function works fine on its own</p>
  
  <pre><code>myFunc(sdt.select('col1'), 'col1')
      min_col1    max_col1    col_name
  0   4   100 col1
  </code></pre>
  
  <p>Issues arise when I put that inside rdd map, similar to what has been done above</p>
  
  <pre><code>x= sdt_col_rdd.map(lambda col: (col,myFunc(sdt.select(col), col)))
  y = x.collectAsMap()
  </code></pre>
  
  <p>Any idea how to achieve this kind of transformations/actions in Spark for columns in parallel, without <em>udaf</em> ? Collect_list would not be efficient because of huge dataset and without exploiting Spark's functions.</p>
  
  <blockquote>
    <p>During handling of the above exception, another exception occurred:</p>
    
    <p>PicklingError                             Traceback (most recent call
    last)  in ()
          1 col_map = sdt_col_rdd.map(lambda col: (col,myFunc(sdt.select(col), col)))
    ----> 2 y = col_map.collectAsMap()</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/rdd.py in collectAsMap(self)    1555         4    1556         """
    -> 1557         return dict(self.collect())    1558     1559     def keys(self):</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/rdd.py in collect(self)
        794         """
        795         with SCCallSiteSync(self.context) as css:
    --> 796             sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
        797         return list(_load_from_socket(sock_info, self._jrdd_deserializer))
        798 </p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/rdd.py in _jrdd(self)    2440     2441         wrapped_func =
    _wrap_function(self.ctx, self.func, self._prev_jrdd_deserializer,
    -> 2442                                       self._jrdd_deserializer, profiler)    2443         python_rdd =
    self.ctx._jvm.PythonRDD(self._prev_jrdd.rdd(), wrapped_func,    2444<br>
    self.preservesPartitioning)</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/rdd.py in _wrap_function(sc, func, deserializer, serializer, profiler)<br>
    2373     assert serializer, "serializer should not be empty"    2374<br>
    command = (func, profiler, deserializer, serializer)
    -> 2375     pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)    2376     return sc._jvm.PythonFunction(bytearray(pickled_command), env, includes,
    sc.pythonExec,    2377                                   sc.pythonVer,
    broadcast_vars, sc._javaAccumulator)</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/rdd.py in _prepare_for_python_RDD(sc, command)    2359     # the serialized
    command will be compressed by broadcast    2360     ser =
    CloudPickleSerializer()
    -> 2361     pickled_command = ser.dumps(command)    2362     if len(pickled_command) > (1 &lt;&lt; 20):  # 1M    2363         # The
    broadcast will have same life cycle as created PythonRDD</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/serializers.py
    in dumps(self, obj)
        462 
        463     def dumps(self, obj):
    --> 464         return cloudpickle.dumps(obj, 2)
        465 
        466 </p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/cloudpickle.py
    in dumps(obj, protocol)
        702 
        703     cp = CloudPickler(file,protocol)
    --> 704     cp.dump(obj)
        705 
        706     return file.getvalue()</p>
    
    <p>/data/2/parcels/SPARK2-2.2.0.cloudera4-1.cdh5.13.3.p0.603055/lib/spark2/python/pyspark/cloudpickle.py
    in dump(self, obj)
        160                 msg = "Could not serialize object: %s: %s" % (e.<strong>class</strong>.<strong>name</strong>, emsg)
        161             print_exec(sys.stderr)
    --> 162             raise pickle.PicklingError(msg)
        163 
        164     def save_memoryview(self, obj):</p>
    
    <p>PicklingError: Could not serialize object: Py4JError: An error
    occurred while calling o62.<strong>getstate</strong>. Trace: py4j.Py4JException:
    Method <strong>getstate</strong>([]) does not exist    at
    py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
      at
    py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
      at py4j.Gateway.invoke(Gateway.java:274)    at
    py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
      at py4j.commands.CallCommand.execute(CallCommand.java:79)   at
    py4j.GatewayConnection.run(GatewayConnection.java:238)    at
    java.lang.Thread.run(Thread.java:748)</p>
  </blockquote>