Return a new PRDD by applying a function to each element of each
        pandas DataFrame.
965833
def applymap(self, func, **kwargs):
        """"""
        return self.from_rdd(
            self._rdd.map(lambda data: data.applymap(func), **kwargs))
212516
def despike(df, n1=2, n2=20, block=100, keep=0):
    
    if isinstance(df, pd.Series):
        new_df = _despike(df, n1=n1, n2=n2, block=block, keep=keep)
    else:
        new_df = df.apply(_despike, n1=n1, n2=n2, block=block, keep=keep)
    return new_df
1078972
def gen_rsd_cdf(K, delta, c):
    """"""

    mu = gen_mu(K, delta, c)
    return [sum(mu[:d+1]) for d in range(K)]
443140
def cols_(df,col_list,func,*args,**kwargs):
    
    return _pd.DataFrame({col_name:func(df,col_name,*args,**kwargs) for col_name in col_list})
1222566
def apply(self, func, *args, **kwargs):
        
        self._prep_pandas_groupby()

        def key_by_index(data):
            """Key each row by its index.
            """
            # TODO: Is there a better way to do this?
            for key, row in data.iterrows():
                yield (key, pd.DataFrame.from_dict(
                    dict([(key, row)]), orient='index'))

        myargs = self._myargs
        mykwargs = self._mykwargs
        regroupedRDD = self._distributedRDD.mapValues(
            lambda data: data.groupby(*myargs, **mykwargs))
        appliedRDD = regroupedRDD.map(
            lambda key_data: key_data[1].apply(func, *args, **kwargs))
        reKeyedRDD = appliedRDD.flatMap(key_by_index)
        dataframe = self._sortIfNeeded(reKeyedRDD).values()
        return DataFrame.fromDataFrameRDD(dataframe, self.sql_ctx)
1083987
def assign_norrec_to_df(df):
    
    if df.shape[0] == 0:
        # empty dataframe, just return a copy
        return df.copy()

    c = df[['a', 'b', 'm', 'n']].values.copy()
    # unique injections
    cu = np.unique(c, axis=0)

    print('generating ids')
    # now assign unique IDs to each config in normal and reciprocal
    running_index = 0
    normal_ids = {}
    reciprocal_ids = {}
    # loop through all configurations
    for i in range(0, cu.shape[0]):
        # print('testing', cu[i], i, cu.shape[0])
        # normalize configuration
        cu_norm = _normalize_abmn(cu[i, :]).squeeze()
        if tuple(cu_norm) in normal_ids:
            # print('already indexed')
            continue

        # find pairs
        indices = np.where((
            # current electrodes
            (
                (
                    (cu[:, 0] == cu[i, 2]) & (cu[:, 1] == cu[i, 3])
                ) |
                (
                    (cu[:, 0] == cu[i, 3]) & (cu[:, 1] == cu[i, 2])
                )
            ) &
            # voltage electrodes
            (
                (
                    (cu[:, 2] == cu[i, 0]) & (cu[:, 3] == cu[i, 1])
                ) |
                (
                    (cu[:, 2] == cu[i, 1]) & (cu[:, 3] == cu[i, 0])
                )
            )
        ))[0]

        # we found no pair
        if len(indices) == 0:
            # print('no reciprocals, continuing')
            if not tuple(cu_norm) in normal_ids:
                if np.min(cu_norm[0:2]) < np.min(cu_norm[2:3]):
                    # treat as normal
                    normal_ids[tuple(cu_norm)] = running_index
                else:
                    reciprocal_ids[tuple(cu_norm)] = running_index
                running_index += 1
            continue

        # if len(indices) > 1:
        #     print('found more than one reciprocals')

        # normalize the first reciprocal
        cu_rec_norm = _normalize_abmn(cu[indices[0], :]).squeeze()

        # decide on normal or reciprocal
        # print('ABREC', cu_norm[0:2], cu_rec_norm[0:2])
        if np.min(cu_norm[0:2]) < np.min(cu_rec_norm[0:2]):
            # print('is normal')
            # normal
            normal_ids[tuple(cu_norm)] = running_index
            reciprocal_ids[tuple(cu_rec_norm)] = running_index
        else:
            normal_ids[tuple(cu_rec_norm)] = running_index
            reciprocal_ids[tuple(cu_norm)] = running_index
        running_index += 1

    print('assigning ids')
    # print(df.shape)
    # print(df.columns)
    # print('normal_ids', normal_ids)
    # print('reciprocal_ids', reciprocal_ids)
    # now convert the indices into a dataframe so we can use pd.merge
    # note that this code was previously written in another way, so the
    # conversion is quite cumbersome
    # at one point we need to rewrite everything here...
    df_nor = {item: key for key, item in normal_ids.items()}
    df_nor = pd.DataFrame(df_nor).T.reset_index().rename(
        {'index': 'id'}, axis=1)
    df_nor['norrec'] = 'nor'

    if len(normal_ids) > 0:
        df_nor.columns = ('id', 'a', 'b', 'm', 'n', 'norrec')
        df_nor2 = df_nor.copy()
        df_nor2.columns = ('id', 'b', 'a', 'm', 'n', 'norrec')
        df_nor3 = df_nor.copy()
        df_nor3.columns = ('id', 'b', 'a', 'n', 'm', 'norrec')
        df_nor4 = df_nor.copy()
        df_nor4.columns = ('id', 'a', 'b', 'n', 'm', 'norrec')
        df_ids = pd.concat(
            (
                df_nor,
                df_nor2,
                df_nor3,
                df_nor4,
            ),
            sort=True
        )
    else:
        df_ids = pd.DataFrame()

    if len(reciprocal_ids) > 0:
        df_rec = {item: key for key, item in reciprocal_ids.items()}
        df_rec = pd.DataFrame(df_rec).T.reset_index().rename(
            {'index': 'id'}, axis=1)
        df_rec['norrec'] = 'rec'
        df_rec.columns = ('id', 'a', 'b', 'm', 'n', 'norrec')
        df_rec2 = df_rec.copy()
        df_rec2.columns = ('id', 'b', 'a', 'm', 'n', 'norrec')
        df_rec3 = df_rec.copy()
        df_rec3.columns = ('id', 'b', 'a', 'n', 'm', 'norrec')
        df_rec4 = df_rec.copy()
        df_rec4.columns = ('id', 'a', 'b', 'n', 'm', 'norrec')

        df_ids = pd.concat(
            (
                df_ids,
                df_rec,
                df_rec2,
                df_rec3,
                df_rec4,
            ),
            sort=True
        )

    df_new = pd.merge(df, df_ids, how='left', on=('a', 'b', 'm', 'n'))
    df_new.rename(
        {'id_y': 'id',
         'norrec_y': 'norrec'
         }, axis=1,
        inplace=True
    )
    return df_new
    import IPython
    IPython.embed()
    exit()

    df_new[['a', 'b', 'm', 'n', 'id_y', 'norrec_y']]
    # x.iloc[[0, 1978], :]

    # now assign to all measurements
    for key, item in normal_ids.items():
        df.loc[
            ((df.a == key[0]) & (df.b == key[1]) &
             (df.m == key[2]) & (df.n == key[3])) |
            ((df.a == key[1]) & (df.b == key[0]) &
             (df.m == key[2]) & (df.n == key[3])) |
            ((df.a == key[0]) & (df.b == key[1]) &
             (df.m == key[3]) & (df.n == key[2])) |
            ((df.a == key[1]) & (df.b == key[0]) &
             (df.m == key[3]) & (df.n == key[2])),
            ('id', 'norrec')
        ] = (item, 'nor')
    for key, item in reciprocal_ids.items():
        df.loc[
            ((df.a == key[0]) & (df.b == key[1]) &
             (df.m == key[2]) & (df.n == key[3])) |
            ((df.a == key[1]) & (df.b == key[0]) &
             (df.m == key[2]) & (df.n == key[3])) |
            ((df.a == key[0]) & (df.b == key[1]) &
             (df.m == key[3]) & (df.n == key[2])) |
            ((df.a == key[1]) & (df.b == key[0]) &
             (df.m == key[3]) & (df.n == key[2])),
            ('id', 'norrec')
        ] = [item, 'rec']

    # cast norrec-column to string
    df['norrec'] = df['norrec'].astype(str)

    return df
86084
def __generate_method(name):
        
        try:
            func = getattr(DataFrame, name)
        except AttributeError as e:
            # PySpark version is too old
            def func(self, *args, **kwargs):
                raise e
            return func
        wraps = getattr(functools, "wraps", lambda _: lambda f: f)  # py3.4+

        @wraps(func)
        def _wrapper(self, *args, **kwargs):
            dataframe = func(self, *args, **kwargs)
            if self.__class__ != SourcedDataFrame \
                    and isinstance(self, SourcedDataFrame) \
                    and isinstance(dataframe, DataFrame):
                return self.__class__(dataframe._jdf, self._session, self._implicits)
            return dataframe

        return _wrapper
1323910
def _split_on_reappear(cls, df, p, id_offset):
        

        next_id = id_offset + 1
        added_ids = []
        nt = p.sum(0)
        start = np.argmax(p, 0)
        end = np.argmax(np.cumsum(p, 0), 0)
        diff = end - start + 1
        is_contiguous = np.equal(nt, diff)
        for id, contiguous in enumerate(is_contiguous):
            if not contiguous:

                to_change = df[df.id == id]
                index = to_change.index
                diff = index[1:] - index[:-1]
                where = np.where(np.greater(diff, 1))[0]
                for w in where:
                    to_change.loc[w + 1:, 'id'] = next_id
                    added_ids.append(next_id)
                    next_id += 1

                df[df.id == id] = to_change

        return added_ids
264506
def avail(df):
    """"""
    avail = DataFrame(
        {
            "start": df.apply(lambda col: col.first_valid_index()),
            "end": df.apply(lambda col: col.last_valid_index()),
        }
    )
    return avail[["start", "end"]]
632063
def broadcast(df):
    """"""

    sc = SparkContext._active_spark_context
    return DataFrame(sc._jvm.functions.broadcast(df._jdf), df.sql_ctx)
605045
def _regroup_mergedRDD(self):
        
        myargs = self._myargs
        mykwargs = self._mykwargs
        self._prep_pandas_groupby()

        def regroup(df):
            return df.groupby(*myargs, **mykwargs)

        return self._mergedRDD.mapValues(regroup)
1161798
def _make_df(rows):
    """"""

    # Make DataFrame
    df = pd.DataFrame(rows).set_index('ts')

    # TimeDelta Support: https://issues.apache.org/jira/browse/ARROW-835
    for column in df.columns:
        if(df[column].dtype == 'timedelta64[ns]'):
            print('Converting timedelta column {:s}...'.format(column))
            df[column] = df[column].astype(str)
    return df
656299
def smooth(df, window_len=11, window="hanning"):
    """"""
    if isinstance(df, pd.Series):
        new_df = _smooth(df, window_len=window_len, window=window)
    else:
        new_df = df.apply(_smooth, window_len=window_len, window=window)
    return new_df
646004
def nth(self, n, *args, **kwargs):
        """"""
        # TODO: Stop collecting the entire frame for each key.
        self._prep_pandas_groupby()
        myargs = self._myargs
        mykwargs = self._mykwargs
        nthRDD = self._regroup_mergedRDD().mapValues(
            lambda r: r.nth(
                n, *args, **kwargs)).values()
        return DataFrame.fromDataFrameRDD(nthRDD, self.sql_ctx)
471335
def dataframe(self):
        
        frames = []
        for game in self.__iter__():
            df = game.dataframe
            if df is not None:
                frames.append(df)
        if frames == []:
            return None
        return pd.concat(frames)
762898
def clean_add_strdec(*args, prec=28):
    """"""

    # load modules
    import pandas as pd
    import numpy as np
    import re
    from decimal import Decimal, getcontext
    getcontext().prec = prec

    # initialize result as 0.0
    def proc_elem(*args):
        t = Decimal('0.0')
        for a in args:
            if isinstance(a, str):
                a = re.sub('[^0-9\.\-]+', '', a)

            if a and pd.notnull(a):
                t += Decimal(a)

        return str(t)

    def proc_list(arr):
        return [proc_elem(*row) for row in arr]

    def proc_ndarray(arr):
        return np.array(proc_list(arr))

    # transform depending on provided datatypes
    if isinstance(args[0], (list, tuple)):
        return proc_list(args[0])
    elif isinstance(args[0], np.ndarray):
        return proc_ndarray(args[0])
    elif isinstance(args[0], pd.DataFrame):
        return pd.DataFrame(proc_ndarray(args[0].values),
                            index=args[0].index)
    else:
        return proc_elem(*args)
1334690
def dataframe(self,asql, logger = None):
        """"""
        import pandas as pd
        from ambry.mprlib.exceptions import BadSQLError

        try:
            def yielder(cursor):

                for i, row in enumerate(cursor):
                    if i == 0:
                        yield [ e[0] for e in cursor.getdescription()]

                    yield row

            cursor = self.query(asql, logger)

            yld = yielder(cursor)

            header = next(yld)

            return pd.DataFrame(yld, columns=header)
        except BadSQLError as e:
            import traceback
            self._logger.error("SQL Error: {}".format( e))
            self._logger.debug(traceback.format_exc())