def gapply(grouped_data, func, schema, *cols):
import pandas as pd<EOL>minPandasVersion = '<STR_LIT>'<EOL>if LooseVersion(pd.__version__) < LooseVersion(minPandasVersion):<EOL><INDENT>raise ImportError('<STR_LIT>'<EOL>.format(pd.__version__, minPandasVersion))<EOL><DEDENT>keySchema = grouped_data.agg({}).schema<EOL>keyCols = grouped_data.agg({}).columns<EOL>if not cols:<EOL><INDENT>javaDFName = "<STR_LIT>"<EOL>parentDF = java_gateway.get_field(grouped_data._jgd, javaDFName)<EOL>allCols = DataFrame(parentDF, None).columns<EOL>keyColsSet = set(keyCols)<EOL>cols = [col for col in allCols if col not in keyColsSet]<EOL><DEDENT>if "<STR_LIT:*>" in cols:<EOL><INDENT>raise ValueError("<STR_LIT>")<EOL><DEDENT>if len(set(cols)) < len(cols):<EOL><INDENT>raise ValueError("<STR_LIT>")<EOL><DEDENT>if not isinstance(schema, StructType):<EOL><INDENT>raise ValueError("<STR_LIT>")<EOL><DEDENT>inputAggDF = grouped_data.agg({col: '<STR_LIT>' for col in cols})<EOL>canonicalOrder = chain(keyCols, [inputAggDF['<STR_LIT>' + col + '<STR_LIT:)>'] for col in cols])<EOL>inputAggDF = inputAggDF.select(*canonicalOrder)<EOL>def pandasWrappedFunc(*args):<EOL><INDENT>nvals = len(cols)<EOL>keys, collectedCols = args[:-nvals], args[-nvals:]<EOL>paramKeys = tuple(keys)<EOL>if len(paramKeys) == <NUM_LIT:1>:<EOL><INDENT>paramKeys = paramKeys[<NUM_LIT:0>]<EOL><DEDENT>valuesDF = pd.DataFrame.from_dict(dict(zip(cols, collectedCols)))<EOL>valuesDF = valuesDF[list(cols)]  <EOL>outputDF = func(paramKeys, valuesDF)<EOL>valCols = outputDF.columns.tolist()<EOL>for key, keyName in zip(keys, keyCols):<EOL><INDENT>outputDF[keyName] = key<EOL><DEDENT>outputDF = outputDF[keyCols + valCols]  <EOL>return outputDF.values.tolist()<EOL><DEDENT>keyPrependedSchema = StructType(list(chain(keySchema, schema)))<EOL>outputAggSchema = ArrayType(keyPrependedSchema, containsNull=False)<EOL>pandasUDF = udf(pandasWrappedFunc, outputAggSchema)<EOL>outputAggDF = inputAggDF.select(pandasUDF(*inputAggDF))<EOL>explodedDF = outputAggDF.select(explode(*outputAggDF).alias("<STR_LIT>"))<EOL>return explodedDF.select("<STR_LIT>")<EOL>
Applies the function ``func`` to data grouped by key. In particular, given a dataframe
    grouped by some set of key columns key1, key2, ..., keyn, this method groups all the values
    for each row with the same key columns into a single Pandas dataframe and by default invokes
    ``func((key1, key2, ..., keyn), values)`` where the number and order of the key arguments is
    determined by columns on which this instance's parent :class:`DataFrame` was grouped and
    ``values`` is a ``pandas.DataFrame`` of columns selected by ``cols``, in that order.

    If there is only one key then the key tuple is automatically unpacked, with
    ``func(key, values)`` called.

    ``func`` is expected to return a ``pandas.DataFrame`` of the specified schema ``schema``,
    which should be of type :class:`StructType` (output columns are of this name and order).

    If ``spark.conf.get("spark.sql.retainGroupColumns")`` is not ``u'true'``, then ``func`` is
    called with an empty key tuple (note it is set to ``u'true'`` by default).

    If no ``cols`` are specified, then all grouped columns will be offered, in the order of the
    columns in the original dataframe. In either case, the Pandas columns will be named
    according to the DataFrame column names.

    The order of the rows passed in as Pandas rows is not guaranteed to be stable relative to
    the original row order.

    :note: Users must ensure that the grouped values for every group must fit entirely in memory.
    :note: This method is only available if Pandas is installed.

    :param grouped_data: data grouped by key
    :param func: a two argument function, which may be either a lambda or named function
    :param schema: the return schema for ``func``, a :class:`StructType`
    :param cols: list of column names (string only)

    :raise ValueError: if ``"*"`` is in ``cols``
    :raise ValueError: if ``cols`` contains duplicates
    :raise ValueError: if ``schema`` is not a :class:`StructType`
    :raise ImportError: if ``pandas`` module is not installed
    :raise ImportError: if ``pandas`` version is too old (less than 0.7.1)

    :return: the new :class:`DataFrame` with the original key columns replicated for each returned
             value in each group's resulting pandas dataframe, the schema being the original key
             schema prepended to ``schema``, where all the resulting groups' rows are concatenated.
             Of course, if retaining group columns is disabled, then the output will exactly match
             ``schema`` since no keys can be prepended.

    >>> import pandas as pd
    >>> from pyspark.sql import SparkSession
    >>> from spark_sklearn.group_apply import gapply
    >>> from spark_sklearn.util import createLocalSparkSession
    >>> spark = createLocalSparkSession()
    >>> df = (spark
    ...     .createDataFrame([Row(course="dotNET", year=2012, earnings=10000),
    ...                       Row(course="Java",   year=2012, earnings=20000),
    ...                       Row(course="dotNET", year=2012, earnings=5000),
    ...                       Row(course="dotNET", year=2013, earnings=48000),
    ...                       Row(course="Java",   year=2013, earnings=30000)])
    ...     .select("course", "year", "earnings"))
    >>> def yearlyMedian(_, vals):
    ...     all_years = set(vals['year'])
    ...     # Note that interpolation is performed, so we need to cast back to int.
    ...     yearly_median = [(year, int(vals['earnings'][vals['year'] == year].median()))
    ...                      for year in all_years]
    ...     return pd.DataFrame.from_records(yearly_median)
    >>> newSchema = StructType().add("year", LongType()).add("median_earnings", LongType())
    >>> gapply(df.groupBy("course"), yearlyMedian, newSchema).orderBy("median_earnings").show()
    +------+----+---------------+
    |course|year|median_earnings|
    +------+----+---------------+
    |dotNET|2012|           7500|
    |  Java|2012|          20000|
    |  Java|2013|          30000|
    |dotNET|2013|          48000|
    +------+----+---------------+
    <BLANKLINE>
    >>> def twoKeyYearlyMedian(_, vals):
    ...     return pd.DataFrame.from_records([(int(vals["earnings"].median()),)])
    >>> newSchema = StructType([df.schema["earnings"]])
    >>> gapply(df.groupBy("course", "year"), twoKeyYearlyMedian, newSchema, "earnings").orderBy(
    ...     "earnings").show()
    +------+----+--------+
    |course|year|earnings|
    +------+----+--------+
    |dotNET|2012|    7500|
    |  Java|2012|   20000|
    |  Java|2013|   30000|
    |dotNET|2013|   48000|
    +------+----+--------+
    <BLANKLINE>
    >>> spark.stop(); SparkSession._instantiatedContext = None