title: How to convert a group and agg code on a dataframe to a udf in PySpark?
body: <p>I am trying to create a udf for my code for generalizing the problem. I run into issues where it seems like I cannot pass a dataframe to the function.</p>
  
  <p>Input DataFrame:</p>
  
  <pre><code>
  df = sqlContext.createDataFrame([('1', 201001,3400,1600,65,320,400,), ('1', 201002,5200,1600,65,320,400,), ('1', 201003,65,1550,32,320,400,), ('2', 201505,3200,1800,12,1,40,), ('2', 201508,3200,3200,12,1,40,), ('3', 201412,40,40,12,1,3,)], 
                                    ['ColA', 'Col1','Col2','Col3','Col4','Col5','Col6',])
  
  +----+------+----+----+----+----+----+
  |ColA|  Col1|Col2|Col3|Col4|Col5|Col6|
  +----+------+----+----+----+----+----+
  |   1|201001|3400|1600|  65| 320| 400|
  |   1|201002|5200|1600|  65| 320| 400|
  |   1|201003|  65|1550|  32| 320| 400|
  |   2|201505|3200|1800|  12|   1|  40|
  |   2|201508|3200|3200|  12|   1|  40|
  |   3|201412|  40|  40|  12|   1|   3|
  +----+------+----+----+----+----+----+
  </code></pre>
  
  <p>Expected Ouput:</p>
  
  <pre><code>df = sqlContext.createDataFrame([(1,['201001', '201002', '201003'],[3400, 5200, 65],[1600, 1600, 1550],[65,32],[320],[400],), (2,['201505', '201508'],[3200, 3200],[1800, 3200],[12],[1],[40],),
  (3,['201412'],[40],[40],[12],[1],[3],)], ['ColA', 'Col1','Col2','Col3','Col4','Col5','Col6',])
  df.show()
  
  +----+--------------------+----------------+------------------+--------+-----+-----+
  |ColA|                Col1|            Col2|              Col3|    Col4| Col5| Col6|
  +----+--------------------+----------------+------------------+--------+-----+-----+
  |   1|[201001, 201002, ...|[3400, 5200, 65]|[1600, 1600, 1550]|[65, 32]|[320]|[400]|
  |   2|    [201505, 201508]|    [3200, 3200]|      [1800, 3200]|    [12]|  [1]| [40]|
  |   3|            [201412]|            [40]|              [40]|    [12]|  [1]|  [3]|
  +----+--------------------+----------------+------------------+--------+-----+-----+
  </code></pre>
  
  <p>This is the code that works (non-functional)</p>
  
  <pre><code>groupBy = ['ColA']
  
  convert_to_list = ['Col1', 'Col2', 'Col3',]
  convert_to_set = ['Col4', 'Col5', 'Col6',]
  
  exprs = [F.collect_set(F.col(c)).alias(c) for c in cols_to_list]\
            + [F.collect_set(F.col(c)).alias(c) in funs_set for c in 
  
  df = df.groupby(*groupBy).agg(*exprs)
  
  </code></pre>
  
  <p>When I try to create a udf, I get this error:</p>
  
  <pre><code>@F.udf
  def aggregation(df, groupby_column, cols_to_list, cols_to_set):
    exprs = [F.collect_set(F.col(c)).alias(c) for c in cols_to_list]\
            + [F.collect_set(F.col(c)).alias(c) in funs_set for c in cols_to_set]
    return df.groupby(*groupby_column).agg(*exprs)
  
  
  groupby_column = ['ColA']
  cols_to_list = ['Col1', 'Col2', 'Col3',]
  cols_to_set = ['Col4', 'Col5', 'Col6',]
  
  exprs = F.concat([f(F.col(c)) for f in fun_list for c in convert_to_list], [f(F.col(c)) for f in funs_set for c in convert_to_set])
  
  df = df.groupby(*groupBy).agg(*exprs)
  
  TypeError: Invalid argument, not a string or column: DataFrame
  
  </code></pre>