How to convert a group and agg code on a dataframe

Question

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.

Input DataFrame:

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|
+----+------+----+----+----+----+----+

Expected Ouput:

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]|
+----+--------------------+----------------+------------------+--------+-----+-----+

This is the code that works (non-functional)

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)

When I try to create a udf, I get this error:

@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