<s> from __future__ import absolute_import , division , print_function <EOL> import bisect <EOL> from collections import Iterator <EOL> from datetime import datetime <EOL> from distutils . version import LooseVersion <EOL> import math <EOL> import operator <EOL> from operator import getitem , setitem <EOL> from pprint import pformat <EOL> import uuid <EOL> import warnings <EOL> from toolz import merge , partial , first , partition , unique <EOL> import pandas as pd <EOL> from pandas . util . decorators import cache_readonly <EOL> import numpy as np <EOL> try : <EOL> from chest import Chest as Cache <EOL> except ImportError : <EOL> Cache = dict <EOL> from . . import array as da <EOL> from . . import core <EOL> from . . array . core import partial_by_order <EOL> from . . import threaded <EOL> from . . compatibility import apply , operator_div , bind_method <EOL> from . . utils import ( repr_long_list , IndexCallable , <EOL> pseudorandom , derived_from , different_seeds ) <EOL> from . . base import Base , compute , tokenize , normalize_token <EOL> from . . async import get_sync <EOL> no_default = '<STR_LIT>' <EOL> return_scalar = '<STR_LIT>' <EOL> pd . computation . expressions . set_use_numexpr ( False ) <EOL> def _concat ( args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not args : <EOL> return args <EOL> if isinstance ( first ( core . flatten ( args ) ) , np . ndarray ) : <EOL> return da . core . concatenate3 ( args ) <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> return args [ <NUM_LIT:0> ] <EOL> if isinstance ( args [ <NUM_LIT:0> ] , ( pd . DataFrame , pd . Series ) ) : <EOL> args2 = [ arg for arg in args if len ( arg ) ] <EOL> if not args2 : <EOL> return args [ <NUM_LIT:0> ] <EOL> return pd . concat ( args2 ) <EOL> if isinstance ( args [ <NUM_LIT:0> ] , ( pd . Index ) ) : <EOL> args = [ arg for arg in args if len ( arg ) ] <EOL> return args [ <NUM_LIT:0> ] . append ( args [ <NUM_LIT:1> : ] ) <EOL> return args <EOL> def optimize ( dsk , keys , ** kwargs ) : <EOL> from . optimize import optimize <EOL> return optimize ( dsk , keys , ** kwargs ) <EOL> def finalize ( results ) : <EOL> return _concat ( results ) <EOL> class Scalar ( Base ) : <EOL> """<STR_LIT>""" <EOL> _optimize = staticmethod ( optimize ) <EOL> _default_get = staticmethod ( threaded . get ) <EOL> _finalize = staticmethod ( finalize ) <EOL> def __init__ ( self , dsk , _name , name = None , divisions = None ) : <EOL> self . dask = dsk <EOL> self . _name = _name <EOL> self . divisions = [ None , None ] <EOL> def __array__ ( self ) : <EOL> return np . asarray ( self . compute ( ) ) <EOL> @ property <EOL> def _args ( self ) : <EOL> return ( self . dask , self . _name ) <EOL> @ property <EOL> def key ( self ) : <EOL> return ( self . _name , <NUM_LIT:0> ) <EOL> def _keys ( self ) : <EOL> return [ self . key ] <EOL> @ classmethod <EOL> def _get_unary_operator ( cls , op ) : <EOL> def f ( self ) : <EOL> name = tokenize ( self ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( op , ( self . _name , <NUM_LIT:0> ) ) } <EOL> return Scalar ( merge ( dsk , self . dask ) , name ) <EOL> return f <EOL> @ classmethod <EOL> def _get_binary_operator ( cls , op , inv = False ) : <EOL> return lambda self , other : _scalar_binary ( op , self , other , inv = inv ) <EOL> def _scalar_binary ( op , a , b , inv = False ) : <EOL> name = '<STR_LIT>' . format ( op . __name__ , tokenize ( a , b ) ) <EOL> dsk = a . dask <EOL> if not isinstance ( b , Base ) : <EOL> pass <EOL> elif isinstance ( b , Scalar ) : <EOL> dsk = merge ( dsk , b . dask ) <EOL> b = ( b . _name , <NUM_LIT:0> ) <EOL> else : <EOL> return NotImplemented <EOL> if inv : <EOL> dsk . update ( { ( name , <NUM_LIT:0> ) : ( op , b , ( a . _name , <NUM_LIT:0> ) ) } ) <EOL> else : <EOL> dsk . update ( { ( name , <NUM_LIT:0> ) : ( op , ( a . _name , <NUM_LIT:0> ) , b ) } ) <EOL> if isinstance ( b , ( pd . Series , pd . DataFrame ) ) : <EOL> return _Frame ( dsk , name , b , [ b . index . min ( ) , b . index . max ( ) ] ) <EOL> else : <EOL> return Scalar ( dsk , name ) <EOL> class _Frame ( Base ) : <EOL> """<STR_LIT>""" <EOL> _optimize = staticmethod ( optimize ) <EOL> _default_get = staticmethod ( threaded . get ) <EOL> _finalize = staticmethod ( finalize ) <EOL> def __new__ ( cls , dsk , _name , metadata , divisions ) : <EOL> if ( np . isscalar ( metadata ) or metadata is None or <EOL> isinstance ( metadata , ( Series , pd . Series ) ) ) : <EOL> return Series ( dsk , _name , metadata , divisions ) <EOL> else : <EOL> return DataFrame ( dsk , _name , metadata , divisions ) <EOL> @ property <EOL> def _constructor_sliced ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ property <EOL> def _constructor ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ property <EOL> def npartitions ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . divisions ) - <NUM_LIT:1> <EOL> @ classmethod <EOL> def _build_pd ( cls , metadata ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( metadata , cls ) : <EOL> _pd = metadata . _pd <EOL> known_dtype = metadata . _known_dtype <EOL> elif isinstance ( metadata , cls . _partition_type ) : <EOL> if isinstance ( metadata , pd . Index ) : <EOL> _pd = metadata [ <NUM_LIT:0> : <NUM_LIT:0> ] <EOL> else : <EOL> _pd = metadata . iloc [ <NUM_LIT:0> : <NUM_LIT:0> ] <EOL> known_dtype = True <EOL> else : <EOL> if np . isscalar ( metadata ) or metadata is None : <EOL> _pd = cls . _partition_type ( [ ] , name = metadata ) <EOL> else : <EOL> _pd = cls . _partition_type ( columns = metadata ) <EOL> known_dtype = False <EOL> return _pd , known_dtype <EOL> @ property <EOL> def _args ( self ) : <EOL> return NotImplementedError <EOL> def __getnewargs__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _args <EOL> def _keys ( self ) : <EOL> return [ ( self . _name , i ) for i in range ( self . npartitions ) ] <EOL> @ property <EOL> def index ( self ) : <EOL> """<STR_LIT>""" <EOL> name = self . _name + '<STR_LIT>' <EOL> dsk = dict ( ( ( name , i ) , ( getattr , key , '<STR_LIT:index>' ) ) <EOL> for i , key in enumerate ( self . _keys ( ) ) ) <EOL> return Index ( merge ( dsk , self . dask ) , name , <EOL> self . _pd . index . name , self . divisions ) <EOL> @ property <EOL> def known_divisions ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . divisions ) > <NUM_LIT:0> and self . divisions [ <NUM_LIT:0> ] is not None <EOL> def get_division ( self , n ) : <EOL> """<STR_LIT>""" <EOL> if <NUM_LIT:0> <= n < self . npartitions : <EOL> name = '<STR_LIT>' % ( str ( n ) , self . _name ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( self . _name , n ) } <EOL> divisions = self . divisions [ n : n + <NUM_LIT:2> ] <EOL> return self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , divisions ) <EOL> else : <EOL> msg = "<STR_LIT>" . format ( self . npartitions ) <EOL> raise ValueError ( msg ) <EOL> def cache ( self , cache = Cache ) : <EOL> """<STR_LIT>""" <EOL> if callable ( cache ) : <EOL> cache = cache ( ) <EOL> name = '<STR_LIT>' + uuid . uuid1 ( ) . hex <EOL> dsk = dict ( ( ( name , i ) , ( setitem , cache , ( tuple , list ( key ) ) , key ) ) <EOL> for i , key in enumerate ( self . _keys ( ) ) ) <EOL> self . _get ( merge ( dsk , self . dask ) , list ( dsk . keys ( ) ) ) <EOL> name = '<STR_LIT>' + self . _name <EOL> dsk2 = dict ( ( ( name , i ) , ( getitem , cache , ( tuple , list ( key ) ) ) ) <EOL> for i , key in enumerate ( self . _keys ( ) ) ) <EOL> return self . _constructor ( dsk2 , name , self . _pd , self . divisions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def drop_duplicates ( self , ** kwargs ) : <EOL> assert all ( k in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) for k in kwargs ) <EOL> chunk = lambda s : s . drop_duplicates ( ** kwargs ) <EOL> return aca ( self , chunk = chunk , aggregate = chunk , columns = self . _pd , <EOL> token = '<STR_LIT>' ) <EOL> def __len__ ( self ) : <EOL> return reduction ( self , len , np . sum , token = '<STR_LIT>' ) . compute ( ) <EOL> def map_partitions ( self , func , columns = no_default , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return map_partitions ( func , columns , self , * args , ** kwargs ) <EOL> def random_split ( self , p , random_state = None ) : <EOL> """<STR_LIT>""" <EOL> seeds = different_seeds ( self . npartitions , random_state ) <EOL> dsk_full = dict ( ( ( self . _name + '<STR_LIT>' , i ) , <EOL> ( pd_split , ( self . _name , i ) , p , seed ) ) <EOL> for i , seed in enumerate ( seeds ) ) <EOL> dsks = [ dict ( ( ( self . _name + '<STR_LIT>' % i , j ) , <EOL> ( getitem , ( self . _name + '<STR_LIT>' , j ) , i ) ) <EOL> for j in range ( self . npartitions ) ) <EOL> for i in range ( len ( p ) ) ] <EOL> return [ type ( self ) ( merge ( self . dask , dsk_full , dsk ) , <EOL> self . _name + '<STR_LIT>' % i , <EOL> self . _pd , self . divisions ) <EOL> for i , dsk in enumerate ( dsks ) ] <EOL> def head ( self , n = <NUM_LIT:5> , compute = True ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' % ( n , self . _name ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( lambda x , n : x . head ( n = n ) , ( self . _name , <NUM_LIT:0> ) , n ) } <EOL> result = self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , self . divisions [ : <NUM_LIT:2> ] ) <EOL> if compute : <EOL> result = result . compute ( ) <EOL> return result <EOL> def tail ( self , n = <NUM_LIT:5> , compute = True ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' % ( n , self . _name ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( lambda x , n : x . tail ( n = n ) , <EOL> ( self . _name , self . npartitions - <NUM_LIT:1> ) , n ) } <EOL> result = self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , self . divisions [ - <NUM_LIT:2> : ] ) <EOL> if compute : <EOL> result = result . compute ( ) <EOL> return result <EOL> def _loc ( self , ind ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( ind , Series ) : <EOL> return self . _loc_series ( ind ) <EOL> if self . known_divisions : <EOL> if isinstance ( ind , slice ) : <EOL> return self . _loc_slice ( ind ) <EOL> else : <EOL> return self . _loc_element ( ind ) <EOL> else : <EOL> return map_partitions ( try_loc , self , self , ind ) <EOL> def _loc_series ( self , ind ) : <EOL> if not self . divisions == ind . divisions : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return map_partitions ( lambda df , ind : df . loc [ ind ] , <EOL> self . _pd , self , ind , token = '<STR_LIT>' ) <EOL> def _loc_element ( self , ind ) : <EOL> name = '<STR_LIT>' % ( str ( ind ) , self . _name ) <EOL> part = _partition_of_index_value ( self . divisions , ind ) <EOL> if ind < self . divisions [ <NUM_LIT:0> ] or ind > self . divisions [ - <NUM_LIT:1> ] : <EOL> raise KeyError ( '<STR_LIT>' % str ( ind ) ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( lambda df : df . loc [ ind : ind ] , ( self . _name , part ) ) } <EOL> return self . _constructor ( merge ( self . dask , dsk ) , name , self , [ ind , ind ] ) <EOL> def _loc_slice ( self , ind ) : <EOL> name = '<STR_LIT>' % ( str ( ind ) , self . _name ) <EOL> assert ind . step in ( None , <NUM_LIT:1> ) <EOL> if ind . start : <EOL> start = _partition_of_index_value ( self . divisions , ind . start ) <EOL> else : <EOL> start = <NUM_LIT:0> <EOL> if ind . stop is not None : <EOL> stop = _partition_of_index_value ( self . divisions , ind . stop ) <EOL> else : <EOL> stop = self . npartitions - <NUM_LIT:1> <EOL> istart = _coerce_loc_index ( self . divisions , ind . start ) <EOL> istop = _coerce_loc_index ( self . divisions , ind . stop ) <EOL> if stop == start : <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( _loc , ( self . _name , start ) , ind . start , ind . stop ) } <EOL> divisions = [ istart , istop ] <EOL> else : <EOL> dsk = merge ( <EOL> { ( name , <NUM_LIT:0> ) : ( _loc , ( self . _name , start ) , ind . start , None ) } , <EOL> dict ( ( ( name , i ) , ( self . _name , start + i ) ) <EOL> for i in range ( <NUM_LIT:1> , stop - start ) ) , <EOL> { ( name , stop - start ) : ( _loc , ( self . _name , stop ) , None , ind . stop ) } ) <EOL> divisions = ( ( max ( istart , self . divisions [ start ] ) <EOL> if ind . start is not None <EOL> else self . divisions [ <NUM_LIT:0> ] , ) + <EOL> self . divisions [ start + <NUM_LIT:1> : stop + <NUM_LIT:1> ] + <EOL> ( min ( istop , self . divisions [ stop + <NUM_LIT:1> ] ) <EOL> if ind . stop is not None <EOL> else self . divisions [ - <NUM_LIT:1> ] , ) ) <EOL> assert len ( divisions ) == len ( dsk ) + <NUM_LIT:1> <EOL> return self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , divisions ) <EOL> @ property <EOL> def loc ( self ) : <EOL> """<STR_LIT>""" <EOL> return IndexCallable ( self . _loc ) <EOL> @ property <EOL> def iloc ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def repartition ( self , divisions = None , npartitions = None , force = False ) : <EOL> """<STR_LIT>""" <EOL> if npartitions is not None : <EOL> if npartitions > self . npartitions : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return repartition_npartitions ( self , npartitions ) <EOL> elif divisions is not None : <EOL> return repartition ( self , divisions , force = force ) <EOL> else : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> def __getstate__ ( self ) : <EOL> return self . __dict__ <EOL> def __setstate__ ( self , dict ) : <EOL> self . __dict__ = dict <EOL> @ derived_from ( pd . Series ) <EOL> def fillna ( self , value ) : <EOL> return self . map_partitions ( self . _partition_type . fillna , value = value ) <EOL> def sample ( self , frac , replace = False , random_state = None ) : <EOL> """<STR_LIT>""" <EOL> if random_state is None : <EOL> random_state = np . random . randint ( np . iinfo ( np . int32 ) . max ) <EOL> name = '<STR_LIT>' + tokenize ( self , frac , replace , random_state ) <EOL> func = getattr ( self . _partition_type , '<STR_LIT>' ) <EOL> seeds = different_seeds ( self . npartitions , random_state ) <EOL> dsk = dict ( ( ( name , i ) , <EOL> ( apply , func , ( tuple , [ ( self . _name , i ) ] ) , <EOL> { '<STR_LIT>' : frac , '<STR_LIT>' : seed , <EOL> '<STR_LIT:replace>' : replace } ) ) <EOL> for i , seed in zip ( range ( self . npartitions ) , seeds ) ) <EOL> return self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , self . divisions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def to_hdf ( self , path_or_buf , key , mode = '<STR_LIT:a>' , append = False , complevel = <NUM_LIT:0> , <EOL> complib = None , fletcher32 = False , get = get_sync , ** kwargs ) : <EOL> from . io import to_hdf <EOL> return to_hdf ( self , path_or_buf , key , mode , append , complevel , complib , <EOL> fletcher32 , get = get , ** kwargs ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def to_csv ( self , filename , get = get_sync , ** kwargs ) : <EOL> from . io import to_csv <EOL> return to_csv ( self , filename , get = get , ** kwargs ) <EOL> def to_imperative ( self ) : <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> return self . to_delayed ( ) <EOL> def to_delayed ( self ) : <EOL> """<STR_LIT>""" <EOL> from . . delayed import Value <EOL> return [ Value ( k , [ self . dask ] ) for k in self . _keys ( ) ] <EOL> @ classmethod <EOL> def _get_unary_operator ( cls , op ) : <EOL> return lambda self : elemwise ( op , self ) <EOL> @ classmethod <EOL> def _get_binary_operator ( cls , op , inv = False ) : <EOL> if inv : <EOL> return lambda self , other : elemwise ( op , other , self ) <EOL> else : <EOL> return lambda self , other : elemwise ( op , self , other ) <EOL> def _aca_agg ( self , token , func , aggfunc = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> @ derived_from ( pd . DataFrame ) <EOL> def sum ( self , axis = None , skipna = True ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . sum ( axis = <NUM_LIT:1> , skipna = skipna ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> return self . _aca_agg ( token = '<STR_LIT>' , func = lambda x : x . sum ( skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def max ( self , axis = None , skipna = True ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . max ( axis = <NUM_LIT:1> , skipna = skipna ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> return self . _aca_agg ( token = '<STR_LIT>' , func = lambda x : x . max ( skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def min ( self , axis = None , skipna = True ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . min ( axis = <NUM_LIT:1> , skipna = skipna ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> return self . _aca_agg ( token = '<STR_LIT>' , func = lambda x : x . min ( skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def count ( self , axis = None ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . count ( axis = <NUM_LIT:1> ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> return self . _aca_agg ( token = '<STR_LIT:count>' , func = lambda x : x . count ( ) , <EOL> aggfunc = lambda x : x . sum ( ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def mean ( self , axis = None , skipna = True ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . mean ( axis = <NUM_LIT:1> , skipna = skipna ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> num = self . _get_numeric_data ( ) <EOL> s = num . sum ( skipna = skipna ) <EOL> n = num . count ( ) <EOL> def f ( s , n ) : <EOL> try : <EOL> return s / n <EOL> except ZeroDivisionError : <EOL> return np . nan <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , tokenize ( s ) ) <EOL> return map_partitions ( f , None , s , n , token = name ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def var ( self , axis = None , skipna = True , ddof = <NUM_LIT:1> ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . var ( axis = <NUM_LIT:1> , skipna = skipna , ddof = ddof ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , ddof ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> num = self . _get_numeric_data ( ) <EOL> x = <NUM_LIT:1.0> * num . sum ( skipna = skipna ) <EOL> x2 = <NUM_LIT:1.0> * ( num ** <NUM_LIT:2> ) . sum ( skipna = skipna ) <EOL> n = num . count ( ) <EOL> def f ( x2 , x , n ) : <EOL> try : <EOL> result = ( x2 / n ) - ( x / n ) ** <NUM_LIT:2> <EOL> if ddof : <EOL> result = result * n / ( n - ddof ) <EOL> return result <EOL> except ZeroDivisionError : <EOL> return np . nan <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , ddof ) <EOL> return map_partitions ( f , None , x2 , x , n , token = name ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def std ( self , axis = None , skipna = True , ddof = <NUM_LIT:1> ) : <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> f = lambda x : x . std ( axis = <NUM_LIT:1> , skipna = skipna , ddof = ddof ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , ddof ) <EOL> return map_partitions ( f , None , self , token = name ) <EOL> else : <EOL> v = self . var ( skipna = skipna , ddof = ddof ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , ddof ) <EOL> return map_partitions ( np . sqrt , None , v , token = name ) <EOL> def quantile ( self , q = <NUM_LIT:0.5> , axis = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> axis = self . _validate_axis ( axis ) <EOL> name = '<STR_LIT>' + tokenize ( self , q , axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> if isinstance ( q , list ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return map_partitions ( pd . DataFrame . quantile , None , self , <EOL> q , axis , token = name ) <EOL> else : <EOL> num = self . _get_numeric_data ( ) <EOL> quantiles = tuple ( quantile ( self [ c ] , q ) for c in num . columns ) <EOL> dask = { } <EOL> dask = merge ( dask , * [ q . dask for q in quantiles ] ) <EOL> qnames = [ ( q . _name , <NUM_LIT:0> ) for q in quantiles ] <EOL> if isinstance ( quantiles [ <NUM_LIT:0> ] , Scalar ) : <EOL> dask [ ( name , <NUM_LIT:0> ) ] = ( pd . Series , ( list , qnames ) , num . columns ) <EOL> divisions = ( min ( num . columns ) , max ( num . columns ) ) <EOL> return Series ( dask , name , None , divisions ) <EOL> else : <EOL> from . multi import _pdconcat <EOL> dask [ ( name , <NUM_LIT:0> ) ] = ( _pdconcat , ( list , qnames ) , <NUM_LIT:1> ) <EOL> return DataFrame ( dask , name , num . columns , <EOL> quantiles [ <NUM_LIT:0> ] . divisions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def describe ( self ) : <EOL> name = '<STR_LIT>' + tokenize ( self ) <EOL> num = self . _get_numeric_data ( ) <EOL> stats = [ num . count ( ) , num . mean ( ) , num . std ( ) , num . min ( ) , <EOL> num . quantile ( [ <NUM_LIT> , <NUM_LIT:0.5> , <NUM_LIT> ] ) , num . max ( ) ] <EOL> stats_names = [ ( s . _name , <NUM_LIT:0> ) for s in stats ] <EOL> def build_partition ( values ) : <EOL> assert len ( values ) == <NUM_LIT:6> <EOL> count , mean , std , min , q , max = values <EOL> part1 = self . _partition_type ( [ count , mean , std , min ] , <EOL> index = [ '<STR_LIT:count>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> q . index = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> part3 = self . _partition_type ( [ max ] , index = [ '<STR_LIT>' ] ) <EOL> return pd . concat ( [ part1 , q , part3 ] ) <EOL> dsk = dict ( ) <EOL> dsk [ ( name , <NUM_LIT:0> ) ] = ( build_partition , ( list , stats_names ) ) <EOL> dsk = merge ( dsk , num . dask , * [ s . dask for s in stats ] ) <EOL> return self . _constructor ( dsk , name , num . _pd , <EOL> divisions = [ None , None ] ) <EOL> def _cum_agg ( self , token , chunk , aggregate , axis , skipna = True , <EOL> chunk_kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , token ) <EOL> return self . map_partitions ( chunk , token = name , ** chunk_kwargs ) <EOL> else : <EOL> name1 = '<STR_LIT>' . format ( self . _token_prefix , token ) <EOL> cumpart = map_partitions ( chunk , self . _pd , self , <EOL> token = name1 , ** chunk_kwargs ) <EOL> name2 = '<STR_LIT>' . format ( self . _token_prefix , token ) <EOL> cumlast = map_partitions ( _take_last , None , cumpart , <EOL> skipna , token = name2 ) <EOL> name = '<STR_LIT>' . format ( self . _token_prefix , token ) <EOL> cname = '<STR_LIT>' . format ( self . _token_prefix , token ) <EOL> dask = { } <EOL> dask [ ( name , <NUM_LIT:0> ) ] = ( cumpart . _name , <NUM_LIT:0> ) <EOL> for i in range ( <NUM_LIT:1> , self . npartitions ) : <EOL> if i == <NUM_LIT:1> : <EOL> dask [ ( cname , i ) ] = ( cumlast . _name , i - <NUM_LIT:1> ) <EOL> else : <EOL> dask [ ( cname , i ) ] = ( aggregate , ( cname , i - <NUM_LIT:1> ) , <EOL> ( cumlast . _name , i - <NUM_LIT:1> ) ) <EOL> dask [ ( name , i ) ] = ( aggregate , ( cumpart . _name , i ) , ( cname , i ) ) <EOL> return self . _constructor ( merge ( dask , cumpart . dask , cumlast . dask ) , <EOL> name , chunk ( self . _pd ) , self . divisions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def cumsum ( self , axis = None , skipna = True ) : <EOL> cumsum = lambda x , ** kwargs : x . cumsum ( ** kwargs ) <EOL> return self . _cum_agg ( '<STR_LIT>' , <EOL> chunk = cumsum , <EOL> aggregate = operator . add , <EOL> axis = axis , skipna = skipna , <EOL> chunk_kwargs = dict ( axis = axis , skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def cumprod ( self , axis = None , skipna = True ) : <EOL> cumprod = lambda x , ** kwargs : x . cumprod ( ** kwargs ) <EOL> return self . _cum_agg ( '<STR_LIT>' , <EOL> chunk = cumprod , <EOL> aggregate = operator . mul , <EOL> axis = axis , skipna = skipna , <EOL> chunk_kwargs = dict ( axis = axis , skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def cummax ( self , axis = None , skipna = True ) : <EOL> def aggregate ( x , y ) : <EOL> if isinstance ( x , ( pd . Series , pd . DataFrame ) ) : <EOL> return x . where ( ( x > y ) | x . isnull ( ) , y , axis = x . ndim - <NUM_LIT:1> ) <EOL> else : <EOL> return x if x > y else y <EOL> cummax = lambda x , ** kwargs : x . cummax ( ** kwargs ) <EOL> return self . _cum_agg ( '<STR_LIT>' , <EOL> chunk = cummax , <EOL> aggregate = aggregate , <EOL> axis = axis , skipna = skipna , <EOL> chunk_kwargs = dict ( axis = axis , skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def cummin ( self , axis = None , skipna = True ) : <EOL> def aggregate ( x , y ) : <EOL> if isinstance ( x , ( pd . Series , pd . DataFrame ) ) : <EOL> return x . where ( ( x < y ) | x . isnull ( ) , y , axis = x . ndim - <NUM_LIT:1> ) <EOL> else : <EOL> return x if x < y else y <EOL> cummin = lambda x , ** kwargs : x . cummin ( ** kwargs ) <EOL> return self . _cum_agg ( '<STR_LIT>' , <EOL> chunk = cummin , <EOL> aggregate = aggregate , <EOL> axis = axis , skipna = skipna , <EOL> chunk_kwargs = dict ( axis = axis , skipna = skipna ) ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def where ( self , cond , other = np . nan ) : <EOL> return map_partitions ( self . _partition_type . where , no_default , <EOL> self , cond , other ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def mask ( self , cond , other = np . nan ) : <EOL> return map_partitions ( self . _partition_type . mask , no_default , <EOL> self , cond , other ) <EOL> @ derived_from ( pd . Series ) <EOL> def append ( self , other ) : <EOL> if isinstance ( other , ( list , dict ) ) : <EOL> msg = "<STR_LIT>" <EOL> raise NotImplementedError ( msg ) <EOL> if not isinstance ( other , _Frame ) : <EOL> from . io import from_pandas <EOL> other = from_pandas ( other , <NUM_LIT:1> ) <EOL> from . multi import _append <EOL> if self . known_divisions and other . known_divisions : <EOL> if self . divisions [ - <NUM_LIT:1> ] < other . divisions [ <NUM_LIT:0> ] : <EOL> divisions = self . divisions [ : - <NUM_LIT:1> ] + other . divisions <EOL> return _append ( self , other , divisions ) <EOL> else : <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> raise ValueError ( msg ) <EOL> else : <EOL> divisions = [ None ] * ( self . npartitions + other . npartitions + <NUM_LIT:1> ) <EOL> return _append ( self , other , divisions ) <EOL> @ classmethod <EOL> def _bind_operator_method ( cls , name , op ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> normalize_token . register ( ( Scalar , _Frame ) , lambda a : a . _name ) <EOL> class Series ( _Frame ) : <EOL> """<STR_LIT>""" <EOL> _partition_type = pd . Series <EOL> _token_prefix = '<STR_LIT>' <EOL> def __new__ ( cls , dsk , _name , name , divisions ) : <EOL> result = object . __new__ ( cls ) <EOL> result . dask = dsk <EOL> result . _name = _name <EOL> result . _pd , result . _known_dtype = cls . _build_pd ( name ) <EOL> result . divisions = tuple ( divisions ) <EOL> return result <EOL> @ property <EOL> def _args ( self ) : <EOL> return ( self . dask , self . _name , self . name , self . divisions ) <EOL> @ property <EOL> def _constructor_sliced ( self ) : <EOL> return Scalar <EOL> @ property <EOL> def _constructor ( self ) : <EOL> return Series <EOL> @ property <EOL> def name ( self ) : <EOL> return self . _pd . name <EOL> @ name . setter <EOL> def name ( self , name ) : <EOL> self . _pd . name = name <EOL> renamed = _rename_dask ( self , name ) <EOL> self . dask . update ( renamed . dask ) <EOL> self . _name = renamed . _name <EOL> @ property <EOL> def ndim ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:1> <EOL> @ property <EOL> def dtype ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _known_dtype : <EOL> return self . dtype <EOL> else : <EOL> self . _pd , self . _known_dtype = self . _build_pd ( self . head ( ) ) <EOL> return self . _pd . dtype <EOL> def __getattr__ ( self , key ) : <EOL> if key == '<STR_LIT>' : <EOL> head = self . head ( ) <EOL> if isinstance ( head . dtype , pd . core . dtypes . CategoricalDtype ) : <EOL> return head . cat <EOL> raise AttributeError ( "<STR_LIT>" % key ) <EOL> @ property <EOL> def column_info ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> return self . name <EOL> @ property <EOL> def nbytes ( self ) : <EOL> return reduction ( self , lambda s : s . nbytes , np . sum , token = '<STR_LIT>' ) <EOL> def __repr__ ( self ) : <EOL> return ( "<STR_LIT>" % <EOL> ( self . __class__ . __name__ , self . _name , <EOL> repr_long_list ( self . divisions ) ) ) <EOL> def __array__ ( self , dtype = None , ** kwargs ) : <EOL> x = np . array ( self . compute ( ) ) <EOL> if dtype and x . dtype != dtype : <EOL> x = x . astype ( dtype ) <EOL> return x <EOL> def __array_wrap__ ( self , array , context = None ) : <EOL> return pd . Series ( array , name = self . name ) <EOL> @ cache_readonly <EOL> def dt ( self ) : <EOL> return DatetimeAccessor ( self ) <EOL> @ cache_readonly <EOL> def str ( self ) : <EOL> return StringAccessor ( self ) <EOL> def quantile ( self , q = <NUM_LIT:0.5> ) : <EOL> """<STR_LIT>""" <EOL> return quantile ( self , q ) <EOL> @ derived_from ( pd . Series ) <EOL> def resample ( self , rule , how = None , closed = None , label = None ) : <EOL> from . tseries . resample import _resample <EOL> return _resample ( self , rule , how = how , closed = closed , label = label ) <EOL> def __getitem__ ( self , key ) : <EOL> if isinstance ( key , Series ) and self . divisions == key . divisions : <EOL> name = '<STR_LIT>' % ( self . _name , key . _name ) <EOL> dsk = dict ( ( ( name , i ) , ( operator . getitem , ( self . _name , i ) , <EOL> ( key . _name , i ) ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> return Series ( merge ( self . dask , key . dask , dsk ) , name , <EOL> self . name , self . divisions ) <EOL> raise NotImplementedError ( ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def _get_numeric_data ( self , how = '<STR_LIT>' , subset = None ) : <EOL> return self <EOL> @ derived_from ( pd . Series ) <EOL> def iteritems ( self ) : <EOL> for i in range ( self . npartitions ) : <EOL> s = self . get_division ( i ) . compute ( ) <EOL> for item in s . iteritems ( ) : <EOL> yield item <EOL> @ classmethod <EOL> def _validate_axis ( cls , axis = <NUM_LIT:0> ) : <EOL> if axis not in ( <NUM_LIT:0> , '<STR_LIT:index>' , None ) : <EOL> raise ValueError ( '<STR_LIT>' . format ( axis ) ) <EOL> return { None : <NUM_LIT:0> , '<STR_LIT:index>' : <NUM_LIT:0> } . get ( axis , axis ) <EOL> def _aca_agg ( self , token , func , aggfunc = None ) : <EOL> """<STR_LIT>""" <EOL> if aggfunc is None : <EOL> aggfunc = func <EOL> return aca ( [ self ] , chunk = func , <EOL> aggregate = lambda x : aggfunc ( pd . Series ( x ) ) , <EOL> columns = return_scalar , token = self . _token_prefix + token ) <EOL> @ derived_from ( pd . Series ) <EOL> def groupby ( self , index , ** kwargs ) : <EOL> from dask . dataframe . groupby import SeriesGroupBy <EOL> return SeriesGroupBy ( self , index , ** kwargs ) <EOL> @ derived_from ( pd . Series ) <EOL> def sum ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . sum ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def max ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . max ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def min ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . min ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def count ( self ) : <EOL> return super ( Series , self ) . count ( ) <EOL> @ derived_from ( pd . Series ) <EOL> def mean ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . mean ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def var ( self , axis = None , ddof = <NUM_LIT:1> , skipna = True ) : <EOL> return super ( Series , self ) . var ( axis = axis , ddof = ddof , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def std ( self , axis = None , ddof = <NUM_LIT:1> , skipna = True ) : <EOL> return super ( Series , self ) . std ( axis = axis , ddof = ddof , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def cumsum ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . cumsum ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def cumprod ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . cumprod ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def cummax ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . cummax ( axis = axis , skipna = skipna ) <EOL> @ derived_from ( pd . Series ) <EOL> def cummin ( self , axis = None , skipna = True ) : <EOL> return super ( Series , self ) . cummin ( axis = axis , skipna = skipna ) <EOL> def unique ( self ) : <EOL> """<STR_LIT>""" <EOL> chunk = lambda x : pd . Series ( pd . Series . unique ( x ) , name = self . name ) <EOL> return aca ( self , chunk = chunk , aggregate = chunk , <EOL> columns = self . name , token = '<STR_LIT>' ) <EOL> @ derived_from ( pd . Series ) <EOL> def nunique ( self ) : <EOL> return self . drop_duplicates ( ) . count ( ) <EOL> @ derived_from ( pd . Series ) <EOL> def value_counts ( self ) : <EOL> chunk = lambda s : s . value_counts ( ) <EOL> if LooseVersion ( pd . __version__ ) > '<STR_LIT>' : <EOL> agg = lambda s : s . groupby ( level = <NUM_LIT:0> ) . sum ( ) . sort_values ( ascending = False ) <EOL> else : <EOL> agg = lambda s : s . groupby ( level = <NUM_LIT:0> ) . sum ( ) . sort ( inplace = False , ascending = False ) <EOL> return aca ( self , chunk = chunk , aggregate = agg , columns = self . name , <EOL> token = '<STR_LIT>' ) <EOL> @ derived_from ( pd . Series ) <EOL> def nlargest ( self , n = <NUM_LIT:5> ) : <EOL> return nlargest ( self , n ) <EOL> @ derived_from ( pd . Series ) <EOL> def isin ( self , other ) : <EOL> return elemwise ( pd . Series . isin , self , list ( other ) ) <EOL> @ derived_from ( pd . Series ) <EOL> def map ( self , arg , na_action = None ) : <EOL> if not ( isinstance ( arg , ( pd . Series , dict ) ) or callable ( arg ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( type ( arg ) ) ) <EOL> name = '<STR_LIT>' + tokenize ( self , arg , na_action ) <EOL> dsk = dict ( ( ( name , i ) , ( pd . Series . map , k , arg , na_action ) ) for i , k in <EOL> enumerate ( self . _keys ( ) ) ) <EOL> dsk . update ( self . dask ) <EOL> return Series ( dsk , name , self . name , self . divisions ) <EOL> @ derived_from ( pd . Series ) <EOL> def astype ( self , dtype ) : <EOL> return map_partitions ( pd . Series . astype , self . name , self , dtype = dtype ) <EOL> @ derived_from ( pd . Series ) <EOL> def dropna ( self ) : <EOL> return self . map_partitions ( pd . Series . dropna ) <EOL> @ derived_from ( pd . Series ) <EOL> def between ( self , left , right , inclusive = True ) : <EOL> return self . map_partitions ( pd . Series . between , left = left , <EOL> right = right , inclusive = inclusive ) <EOL> @ derived_from ( pd . Series ) <EOL> def clip ( self , lower = None , upper = None ) : <EOL> return self . map_partitions ( pd . Series . clip , lower = lower , upper = upper ) <EOL> @ derived_from ( pd . Series ) <EOL> def notnull ( self ) : <EOL> return self . map_partitions ( pd . Series . notnull ) <EOL> @ derived_from ( pd . Series ) <EOL> def isnull ( self ) : <EOL> return self . map_partitions ( pd . Series . isnull ) <EOL> def to_bag ( self , index = False ) : <EOL> """<STR_LIT>""" <EOL> from . io import to_bag <EOL> return to_bag ( self , index ) <EOL> @ derived_from ( pd . Series ) <EOL> def to_frame ( self , name = None ) : <EOL> return map_partitions ( pd . Series . to_frame , self . _pd . to_frame ( name ) , self , name ) <EOL> @ classmethod <EOL> def _bind_operator_method ( cls , name , op ) : <EOL> """<STR_LIT>""" <EOL> def meth ( self , other , level = None , fill_value = None , axis = <NUM_LIT:0> ) : <EOL> if not level is None : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> return map_partitions ( op , self . _pd , self , other , <EOL> axis = axis , fill_value = fill_value ) <EOL> meth . __doc__ = op . __doc__ <EOL> bind_method ( cls , name , meth ) <EOL> def apply ( self , func , convert_dtype = True , name = no_default , args = ( ) , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> if name is no_default : <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> warnings . warn ( msg ) <EOL> name = _emulate ( pd . Series . apply , self . head ( ) , func , <EOL> convert_dtype = convert_dtype , <EOL> args = args , ** kwds ) <EOL> return map_partitions ( pd . Series . apply , name , self , func , <EOL> convert_dtype , args , ** kwds ) <EOL> @ derived_from ( pd . Series ) <EOL> def cov ( self , other , min_periods = None ) : <EOL> from . multi import concat <EOL> if not isinstance ( other , Series ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> df = concat ( [ self , other ] , axis = <NUM_LIT:1> ) <EOL> return cov_corr ( df , min_periods , scalar = True ) <EOL> @ derived_from ( pd . Series ) <EOL> def corr ( self , other , method = '<STR_LIT>' , min_periods = None ) : <EOL> from . multi import concat <EOL> if not isinstance ( other , Series ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if method != '<STR_LIT>' : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> df = concat ( [ self , other ] , axis = <NUM_LIT:1> ) <EOL> return cov_corr ( df , min_periods , corr = True , scalar = True ) <EOL> class Index ( Series ) : <EOL> _partition_type = pd . Index <EOL> _token_prefix = '<STR_LIT>' <EOL> @ property <EOL> def index ( self ) : <EOL> msg = "<STR_LIT>" <EOL> raise AttributeError ( msg . format ( self . __class__ . __name__ ) ) <EOL> @ property <EOL> def _constructor ( self ) : <EOL> return Index <EOL> def head ( self , n = <NUM_LIT:5> , compute = True ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' % ( n , self . _name ) <EOL> dsk = { ( name , <NUM_LIT:0> ) : ( lambda x , n : x [ : n ] , ( self . _name , <NUM_LIT:0> ) , n ) } <EOL> result = self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> self . _pd , self . divisions [ : <NUM_LIT:2> ] ) <EOL> if compute : <EOL> result = result . compute ( ) <EOL> return result <EOL> def nunique ( self ) : <EOL> return self . drop_duplicates ( ) . count ( ) <EOL> @ derived_from ( pd . Index ) <EOL> def max ( self ) : <EOL> return self . _aca_agg ( token = '<STR_LIT>' , func = lambda x : x . max ( ) ) <EOL> @ derived_from ( pd . Index ) <EOL> def min ( self ) : <EOL> return self . _aca_agg ( token = '<STR_LIT>' , func = lambda x : x . min ( ) ) <EOL> def count ( self ) : <EOL> f = lambda x : pd . notnull ( x ) . sum ( ) <EOL> return reduction ( self , f , np . sum , token = '<STR_LIT>' ) <EOL> class DataFrame ( _Frame ) : <EOL> """<STR_LIT>""" <EOL> _partition_type = pd . DataFrame <EOL> _token_prefix = '<STR_LIT>' <EOL> def __new__ ( cls , dask , name , columns , divisions ) : <EOL> result = object . __new__ ( cls ) <EOL> result . dask = dask <EOL> result . _name = name <EOL> result . _pd , result . _known_dtype = cls . _build_pd ( columns ) <EOL> result . divisions = tuple ( divisions ) <EOL> return result <EOL> @ property <EOL> def _args ( self ) : <EOL> return ( self . dask , self . _name , self . columns , self . divisions ) <EOL> @ property <EOL> def _constructor_sliced ( self ) : <EOL> return Series <EOL> @ property <EOL> def _constructor ( self ) : <EOL> return DataFrame <EOL> @ property <EOL> def columns ( self ) : <EOL> return self . _pd . columns <EOL> @ columns . setter <EOL> def columns ( self , columns ) : <EOL> self . _pd . columns = columns <EOL> renamed = _rename_dask ( self , columns ) <EOL> self . dask . update ( renamed . dask ) <EOL> self . _name = renamed . _name <EOL> def __getitem__ ( self , key ) : <EOL> if np . isscalar ( key ) : <EOL> dummy = self . _pd [ _extract_pd ( key ) ] <EOL> name = '<STR_LIT>' . format ( self . _name , key ) <EOL> dsk = dict ( ( ( name , i ) , ( operator . getitem , ( self . _name , i ) , key ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> return self . _constructor_sliced ( merge ( self . dask , dsk ) , name , <EOL> dummy , self . divisions ) <EOL> if isinstance ( key , list ) : <EOL> dummy = self . _pd [ _extract_pd ( key ) ] <EOL> name = '<STR_LIT>' % ( self . _name , str ( key ) ) <EOL> dsk = dict ( ( ( name , i ) , ( operator . getitem , <EOL> ( self . _name , i ) , ( list , key ) ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> return self . _constructor ( merge ( self . dask , dsk ) , name , <EOL> dummy , self . divisions ) <EOL> if isinstance ( key , Series ) : <EOL> if self . divisions != key . divisions : <EOL> from . multi import _maybe_align_partitions <EOL> self , key = _maybe_align_partitions ( [ self , key ] ) <EOL> name = '<STR_LIT>' % ( self . _name , key . _name ) <EOL> dsk = dict ( ( ( name , i ) , ( self . _partition_type . _getitem_array , <EOL> ( self . _name , i ) , <EOL> ( key . _name , i ) ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> return self . _constructor ( merge ( self . dask , key . dask , dsk ) , name , <EOL> self , self . divisions ) <EOL> raise NotImplementedError ( key ) <EOL> def __getattr__ ( self , key ) : <EOL> try : <EOL> return self [ key ] <EOL> except KeyError as e : <EOL> raise AttributeError ( e ) <EOL> def __dir__ ( self ) : <EOL> return sorted ( set ( dir ( type ( self ) ) + list ( self . __dict__ ) + <EOL> list ( self . columns ) ) ) <EOL> def __repr__ ( self ) : <EOL> return ( "<STR_LIT>" % <EOL> ( self . _name , repr_long_list ( self . divisions ) ) ) <EOL> @ property <EOL> def ndim ( self ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:2> <EOL> @ property <EOL> def dtypes ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _known_dtype : <EOL> return self . _pd . dtypes <EOL> else : <EOL> self . _pd , self . _known_dtype = self . _build_pd ( self . head ( ) ) <EOL> return self . _pd . dtypes <EOL> @ derived_from ( pd . DataFrame ) <EOL> def set_index ( self , other , drop = True , ** kwargs ) : <EOL> from . shuffle import set_index <EOL> return set_index ( self , other , drop = drop , ** kwargs ) <EOL> def set_partition ( self , column , divisions , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> from . shuffle import set_partition <EOL> return set_partition ( self , column , divisions , ** kwargs ) <EOL> @ property <EOL> def column_info ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> return self . columns <EOL> @ derived_from ( pd . DataFrame ) <EOL> def nlargest ( self , n = <NUM_LIT:5> , columns = None ) : <EOL> return nlargest ( self , n , columns ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def reset_index ( self ) : <EOL> out = self . map_partitions ( self . _partition_type . reset_index ) <EOL> out . divisions = [ None ] * ( self . npartitions + <NUM_LIT:1> ) <EOL> return out <EOL> @ derived_from ( pd . DataFrame ) <EOL> def groupby ( self , key , ** kwargs ) : <EOL> from dask . dataframe . groupby import DataFrameGroupBy <EOL> return DataFrameGroupBy ( self , key , ** kwargs ) <EOL> def categorize ( self , columns = None , ** kwargs ) : <EOL> from dask . dataframe . categorical import categorize <EOL> return categorize ( self , columns , ** kwargs ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def assign ( self , ** kwargs ) : <EOL> pairs = list ( sum ( kwargs . items ( ) , ( ) ) ) <EOL> df2 = self . _pd . assign ( ** _extract_pd ( kwargs ) ) <EOL> return elemwise ( _assign , self , * pairs , columns = df2 ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def rename ( self , index = None , columns = None ) : <EOL> if index is not None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return map_partitions ( pd . DataFrame . rename , no_default , self , <EOL> None , columns ) <EOL> def query ( self , expr , ** kwargs ) : <EOL> """<STR_LIT>""" + pd . DataFrame . query . __doc__ <EOL> name = '<STR_LIT>' % ( self . _name , expr ) <EOL> if kwargs : <EOL> name = name + '<STR_LIT>' + tokenize ( kwargs ) <EOL> dsk = dict ( ( ( name , i ) , ( apply , pd . DataFrame . query , <EOL> ( ( self . _name , i ) , ( expr , ) , kwargs ) ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> else : <EOL> dsk = dict ( ( ( name , i ) , ( pd . DataFrame . query , ( self . _name , i ) , expr ) ) <EOL> for i in range ( self . npartitions ) ) <EOL> dummy = self . _pd . query ( expr , ** kwargs ) <EOL> return self . _constructor ( merge ( dsk , self . dask ) , name , <EOL> dummy , self . divisions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def dropna ( self , how = '<STR_LIT>' , subset = None ) : <EOL> def f ( df , how = how , subset = subset ) : <EOL> return df . dropna ( how = how , subset = subset ) <EOL> return self . map_partitions ( f , how = how , subset = subset ) <EOL> def to_castra ( self , fn = None , categories = None , sorted_index_column = None , <EOL> compute = True , get = get_sync ) : <EOL> """<STR_LIT>""" <EOL> from . io import to_castra <EOL> return to_castra ( self , fn , categories , sorted_index_column , <EOL> compute = compute , get = get ) <EOL> def to_bag ( self , index = False ) : <EOL> """<STR_LIT>""" <EOL> from . io import to_bag <EOL> return to_bag ( self , index ) <EOL> def _get_numeric_data ( self , how = '<STR_LIT>' , subset = None ) : <EOL> if not self . _known_dtype : <EOL> self . dtypes <EOL> numerics = self . _pd . _get_numeric_data ( ) <EOL> if len ( numerics . columns ) < len ( self . columns ) : <EOL> name = self . _token_prefix + '<STR_LIT>' <EOL> return map_partitions ( pd . DataFrame . _get_numeric_data , <EOL> numerics , self , token = name ) <EOL> else : <EOL> return self <EOL> @ classmethod <EOL> def _validate_axis ( cls , axis = <NUM_LIT:0> ) : <EOL> if axis not in ( <NUM_LIT:0> , <NUM_LIT:1> , '<STR_LIT:index>' , '<STR_LIT>' , None ) : <EOL> raise ValueError ( '<STR_LIT>' . format ( axis ) ) <EOL> return { None : <NUM_LIT:0> , '<STR_LIT:index>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:1> } . get ( axis , axis ) <EOL> def _aca_agg ( self , token , func , aggfunc = None ) : <EOL> """<STR_LIT>""" <EOL> if aggfunc is None : <EOL> aggfunc = func <EOL> return aca ( [ self ] , chunk = func , <EOL> aggregate = lambda x : x . groupby ( level = <NUM_LIT:0> ) . apply ( aggfunc ) , <EOL> columns = None , token = self . _token_prefix + token ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def drop ( self , labels , axis = <NUM_LIT:0> ) : <EOL> if axis != <NUM_LIT:1> : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> return elemwise ( pd . DataFrame . drop , self , labels , axis ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def merge ( self , right , how = '<STR_LIT>' , on = None , left_on = None , right_on = None , <EOL> left_index = False , right_index = False , <EOL> suffixes = ( '<STR_LIT>' , '<STR_LIT>' ) , npartitions = None ) : <EOL> if not isinstance ( right , ( DataFrame , pd . DataFrame ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> from . multi import merge <EOL> return merge ( self , right , how = how , on = on , <EOL> left_on = left_on , right_on = right_on , <EOL> left_index = left_index , right_index = right_index , <EOL> suffixes = suffixes , npartitions = npartitions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def join ( self , other , on = None , how = '<STR_LIT:left>' , <EOL> lsuffix = '<STR_LIT>' , rsuffix = '<STR_LIT>' , npartitions = None ) : <EOL> if not isinstance ( other , ( DataFrame , pd . DataFrame ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> from . multi import merge <EOL> return merge ( self , other , how = how , <EOL> left_index = on is None , right_index = True , <EOL> left_on = on , suffixes = [ lsuffix , rsuffix ] , <EOL> npartitions = npartitions ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def append ( self , other ) : <EOL> if isinstance ( other , Series ) : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise ValueError ( msg ) <EOL> elif isinstance ( other , pd . Series ) : <EOL> other = other . to_frame ( ) . T <EOL> return super ( DataFrame , self ) . append ( other ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def iterrows ( self ) : <EOL> for i in range ( self . npartitions ) : <EOL> df = self . get_division ( i ) . compute ( ) <EOL> for row in df . iterrows ( ) : <EOL> yield row <EOL> @ derived_from ( pd . DataFrame ) <EOL> def itertuples ( self ) : <EOL> for i in range ( self . npartitions ) : <EOL> df = self . get_division ( i ) . compute ( ) <EOL> for row in df . itertuples ( ) : <EOL> yield row <EOL> @ classmethod <EOL> def _bind_operator_method ( cls , name , op ) : <EOL> """<STR_LIT>""" <EOL> def meth ( self , other , axis = '<STR_LIT>' , level = None , fill_value = None ) : <EOL> if level is not None : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:1> : <EOL> if isinstance ( other , Series ) : <EOL> msg = '<STR_LIT>' . format ( name ) <EOL> raise ValueError ( msg ) <EOL> dummy = _emulate ( op , self , other , axis = axis , fill_value = fill_value ) <EOL> return map_partitions ( op , dummy , self , other , <EOL> axis = axis , fill_value = fill_value ) <EOL> meth . __doc__ = op . __doc__ <EOL> bind_method ( cls , name , meth ) <EOL> def apply ( self , func , axis = <NUM_LIT:0> , args = ( ) , columns = no_default , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> axis = self . _validate_axis ( axis ) <EOL> if axis == <NUM_LIT:0> : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if columns is no_default : <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> warnings . warn ( msg ) <EOL> columns = _emulate ( pd . DataFrame . apply , self . head ( ) , func , <EOL> axis = axis , args = args , ** kwds ) <EOL> return map_partitions ( pd . DataFrame . apply , columns , self , func , axis , <EOL> False , False , None , args , ** kwds ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def cov ( self , min_periods = None ) : <EOL> return cov_corr ( self , min_periods ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def corr ( self , method = '<STR_LIT>' , min_periods = None ) : <EOL> if method != '<STR_LIT>' : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return cov_corr ( self , min_periods , True ) <EOL> @ derived_from ( pd . DataFrame ) <EOL> def astype ( self , dtype ) : <EOL> empty = self . _pd . astype ( dtype ) <EOL> return map_partitions ( pd . DataFrame . astype , empty , self , dtype = dtype ) <EOL> for op in [ operator . abs , operator . add , operator . and_ , operator_div , <EOL> operator . eq , operator . gt , operator . ge , operator . inv , <EOL> operator . lt , operator . le , operator . mod , operator . mul , <EOL> operator . ne , operator . neg , operator . or_ , operator . pow , <EOL> operator . sub , operator . truediv , operator . floordiv , operator . xor ] : <EOL> _Frame . _bind_operator ( op ) <EOL> Scalar . _bind_operator ( op ) <EOL> for name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> meth = getattr ( pd . DataFrame , name ) <EOL> DataFrame . _bind_operator_method ( name , meth ) <EOL> meth = getattr ( pd . Series , name ) <EOL> Series . _bind_operator_method ( name , meth ) <EOL> def elemwise_property ( attr , s ) : <EOL> return map_partitions ( getattr , s . name , s , attr ) <EOL> for name in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> setattr ( Index , name , property ( partial ( elemwise_property , name ) ) ) <EOL> def nlargest ( df , n = <NUM_LIT:5> , columns = None ) : <EOL> if isinstance ( df , Index ) : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> elif isinstance ( df , Series ) : <EOL> token = '<STR_LIT>' . format ( n ) <EOL> f = lambda s : s . nlargest ( n ) <EOL> elif isinstance ( df , DataFrame ) : <EOL> token = '<STR_LIT>' . format ( n ) <EOL> f = lambda df : df . nlargest ( n , columns ) <EOL> columns = df . columns <EOL> return aca ( df , f , f , columns = columns , token = token ) <EOL> def _assign ( df , * pairs ) : <EOL> kwargs = dict ( partition ( <NUM_LIT:2> , pairs ) ) <EOL> return df . assign ( ** kwargs ) <EOL> def _partition_of_index_value ( divisions , val ) : <EOL> """<STR_LIT>""" <EOL> if divisions [ <NUM_LIT:0> ] is None : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> val = _coerce_loc_index ( divisions , val ) <EOL> i = bisect . bisect_right ( divisions , val ) <EOL> return min ( len ( divisions ) - <NUM_LIT:2> , max ( <NUM_LIT:0> , i - <NUM_LIT:1> ) ) <EOL> def _loc ( df , start , stop , include_right_boundary = True ) : <EOL> """<STR_LIT>""" <EOL> result = df . loc [ start : stop ] <EOL> if not include_right_boundary : <EOL> right_index = result . index . get_slice_bound ( stop , '<STR_LIT:left>' , '<STR_LIT>' ) <EOL> result = result . iloc [ : right_index ] <EOL> return result <EOL> def _coerce_loc_index ( divisions , o ) : <EOL> """<STR_LIT>""" <EOL> if divisions and isinstance ( divisions [ <NUM_LIT:0> ] , datetime ) : <EOL> return pd . Timestamp ( o ) <EOL> if divisions and isinstance ( divisions [ <NUM_LIT:0> ] , np . datetime64 ) : <EOL> return np . datetime64 ( o ) . astype ( divisions [ <NUM_LIT:0> ] . dtype ) <EOL> return o <EOL> def elemwise ( op , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> columns = kwargs . pop ( '<STR_LIT>' , no_default ) <EOL> _name = '<STR_LIT>' + tokenize ( op , kwargs , * args ) <EOL> args = _maybe_from_pandas ( args ) <EOL> from . multi import _maybe_align_partitions <EOL> args = _maybe_align_partitions ( args ) <EOL> dasks = [ arg for arg in args if isinstance ( arg , ( _Frame , Scalar ) ) ] <EOL> dfs = [ df for df in dasks if isinstance ( df , _Frame ) ] <EOL> divisions = dfs [ <NUM_LIT:0> ] . divisions <EOL> n = len ( divisions ) - <NUM_LIT:1> <EOL> other = [ ( i , arg ) for i , arg in enumerate ( args ) <EOL> if not isinstance ( arg , ( _Frame , Scalar ) ) ] <EOL> if other : <EOL> op2 = partial_by_order ( op , other ) <EOL> else : <EOL> op2 = op <EOL> keys = [ d . _keys ( ) * n if isinstance ( d , Scalar ) <EOL> else d . _keys ( ) for d in dasks ] <EOL> dsk = dict ( ( ( _name , i ) , ( op2 , ) + frs ) for i , frs in enumerate ( zip ( * keys ) ) ) <EOL> dsk = merge ( dsk , * [ d . dask for d in dasks ] ) <EOL> if columns is no_default : <EOL> if len ( dfs ) >= <NUM_LIT:2> and len ( dasks ) != len ( dfs ) : <EOL> msg = '<STR_LIT>' <EOL> raise NotImplementedError ( msg ) <EOL> columns = _emulate ( op , * args , ** kwargs ) <EOL> return _Frame ( dsk , _name , columns , divisions ) <EOL> def remove_empties ( seq ) : <EOL> """<STR_LIT>""" <EOL> if not seq : <EOL> return seq <EOL> seq2 = [ x for x in seq <EOL> if not ( isinstance ( x , tuple ) and x and x [ <NUM_LIT:0> ] == '<STR_LIT>' ) ] <EOL> if seq2 : <EOL> return seq2 <EOL> else : <EOL> return [ seq [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ] <EOL> def empty_safe ( func , arg ) : <EOL> """<STR_LIT>""" <EOL> if len ( arg ) == <NUM_LIT:0> : <EOL> return ( '<STR_LIT>' , func ( arg ) ) <EOL> else : <EOL> return func ( arg ) <EOL> def reduction ( x , chunk , aggregate , token = None ) : <EOL> """<STR_LIT>""" <EOL> token_key = tokenize ( x , token or ( chunk , aggregate ) ) <EOL> token = token or '<STR_LIT>' <EOL> a = '<STR_LIT>' . format ( token , token_key ) <EOL> dsk = dict ( ( ( a , i ) , ( empty_safe , chunk , ( x . _name , i ) ) ) <EOL> for i in range ( x . npartitions ) ) <EOL> b = '<STR_LIT>' . format ( token , token_key ) <EOL> dsk2 = { ( b , <NUM_LIT:0> ) : ( aggregate , ( remove_empties , <EOL> [ ( a , i ) for i in range ( x . npartitions ) ] ) ) } <EOL> return Scalar ( merge ( x . dask , dsk , dsk2 ) , b ) <EOL> def _maybe_from_pandas ( dfs ) : <EOL> from . io import from_pandas <EOL> dfs = [ from_pandas ( df , <NUM_LIT:1> ) if isinstance ( df , ( pd . Series , pd . DataFrame ) ) <EOL> else df for df in dfs ] <EOL> return dfs <EOL> def apply_concat_apply ( args , chunk = None , aggregate = None , <EOL> columns = no_default , token = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( args , ( tuple , list ) ) : <EOL> args = [ args ] <EOL> assert all ( arg . npartitions == args [ <NUM_LIT:0> ] . npartitions <EOL> for arg in args if isinstance ( arg , _Frame ) ) <EOL> token_key = tokenize ( token or ( chunk , aggregate ) , columns , * args ) <EOL> token = token or '<STR_LIT>' <EOL> a = '<STR_LIT>' . format ( token , token_key ) <EOL> if len ( args ) == <NUM_LIT:1> and isinstance ( args [ <NUM_LIT:0> ] , _Frame ) : <EOL> dsk = dict ( ( ( a , i ) , ( chunk , key ) ) <EOL> for i , key in enumerate ( args [ <NUM_LIT:0> ] . _keys ( ) ) ) <EOL> else : <EOL> dsk = dict ( ( ( a , i ) , ( apply , chunk , [ ( x . _name , i ) <EOL> if isinstance ( x , _Frame ) <EOL> else x for x in args ] ) ) <EOL> for i in range ( args [ <NUM_LIT:0> ] . npartitions ) ) <EOL> b = '<STR_LIT>' . format ( token , token_key ) <EOL> dsk2 = { ( b , <NUM_LIT:0> ) : ( aggregate , <EOL> ( _concat , <EOL> ( list , [ ( a , i ) for i in range ( args [ <NUM_LIT:0> ] . npartitions ) ] ) ) ) } <EOL> if columns is no_default : <EOL> return_type = type ( args [ <NUM_LIT:0> ] ) <EOL> columns = None <EOL> else : <EOL> return_type = _get_return_type ( args [ <NUM_LIT:0> ] , columns ) <EOL> dasks = [ a . dask for a in args if isinstance ( a , _Frame ) ] <EOL> return return_type ( merge ( dsk , dsk2 , * dasks ) , b , columns , [ None , None ] ) <EOL> aca = apply_concat_apply <EOL> def _get_return_type ( arg , metadata ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( metadata , _Frame ) : <EOL> metadata = metadata . _pd <EOL> if isinstance ( metadata , pd . Series ) : <EOL> return Series <EOL> elif isinstance ( metadata , pd . DataFrame ) : <EOL> return DataFrame <EOL> elif isinstance ( metadata , pd . Index ) and isinstance ( arg , Index ) : <EOL> return Index <EOL> if np . isscalar ( metadata ) or metadata is None : <EOL> if metadata == return_scalar : <EOL> return Scalar <EOL> elif isinstance ( arg , Index ) : <EOL> return Index <EOL> else : <EOL> return Series <EOL> else : <EOL> return DataFrame <EOL> def _extract_pd ( x ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( x , _Frame ) : <EOL> return x . _pd <EOL> elif isinstance ( x , list ) : <EOL> return [ _extract_pd ( _x ) for _x in x ] <EOL> elif isinstance ( x , tuple ) : <EOL> return tuple ( [ _extract_pd ( _x ) for _x in x ] ) <EOL> elif isinstance ( x , dict ) : <EOL> res = { } <EOL> for k in x : <EOL> res [ k ] = _extract_pd ( x [ k ] ) <EOL> return res <EOL> else : <EOL> return x <EOL> def _emulate ( func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return func ( * _extract_pd ( args ) , ** _extract_pd ( kwargs ) ) <EOL> def map_partitions ( func , metadata , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> metadata = _extract_pd ( metadata ) <EOL> assert callable ( func ) <EOL> token = kwargs . pop ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> token_key = tokenize ( token or func , metadata , kwargs , * args ) <EOL> name = '<STR_LIT>' . format ( token , token_key ) <EOL> if all ( isinstance ( arg , Scalar ) for arg in args ) : <EOL> dask = { ( name , <NUM_LIT:0> ) : <EOL> ( apply , func , ( tuple , [ ( arg . _name , <NUM_LIT:0> ) for arg in args ] ) , kwargs ) } <EOL> return Scalar ( merge ( dask , * [ arg . dask for arg in args ] ) , name ) <EOL> args = _maybe_from_pandas ( args ) <EOL> from . multi import _maybe_align_partitions <EOL> args = _maybe_align_partitions ( args ) <EOL> dfs = [ df for df in args if isinstance ( df , _Frame ) ] <EOL> if metadata is no_default : <EOL> try : <EOL> metadata = _emulate ( func , * args , ** kwargs ) <EOL> except Exception : <EOL> metadata = None <EOL> if isinstance ( metadata , pd . DataFrame ) : <EOL> columns = metadata . columns <EOL> elif isinstance ( metadata , pd . Series ) : <EOL> columns = metadata . name <EOL> else : <EOL> columns = metadata <EOL> return_type = _get_return_type ( dfs [ <NUM_LIT:0> ] , metadata ) <EOL> dsk = { } <EOL> for i in range ( dfs [ <NUM_LIT:0> ] . npartitions ) : <EOL> values = [ ( arg . _name , i if isinstance ( arg , _Frame ) else <NUM_LIT:0> ) <EOL> if isinstance ( arg , ( _Frame , Scalar ) ) else arg for arg in args ] <EOL> values = ( apply , func , ( tuple , values ) , kwargs ) <EOL> dsk [ ( name , i ) ] = ( _rename , columns , values ) <EOL> dasks = [ arg . dask for arg in args if isinstance ( arg , ( _Frame , Scalar ) ) ] <EOL> return return_type ( merge ( dsk , * dasks ) , name , metadata , args [ <NUM_LIT:0> ] . divisions ) <EOL> def _rename ( columns , df ) : <EOL> """<STR_LIT>""" <EOL> assert not isinstance ( df , _Frame ) <EOL> if isinstance ( columns , Iterator ) : <EOL> columns = list ( columns ) <EOL> if columns is no_default : <EOL> return df <EOL> if isinstance ( df , pd . DataFrame ) : <EOL> if isinstance ( columns , pd . DataFrame ) : <EOL> columns = columns . columns <EOL> columns = pd . Index ( columns ) <EOL> if len ( columns ) == len ( df . columns ) : <EOL> if columns . equals ( df . columns ) : <EOL> return df <EOL> return df . rename ( columns = dict ( zip ( df . columns , columns ) ) ) <EOL> elif isinstance ( df , ( pd . Series , pd . Index ) ) : <EOL> if isinstance ( columns , ( pd . Series , pd . Index ) ) : <EOL> columns = columns . name <EOL> if name == columns : <EOL> return df <EOL> return pd . Series ( df , name = columns ) <EOL> return df <EOL> def _rename_dask ( df , metadata ) : <EOL> """<STR_LIT>""" <EOL> assert isinstance ( df , _Frame ) <EOL> metadata , _ = df . _build_pd ( metadata ) <EOL> name = '<STR_LIT>' . format ( tokenize ( df , metadata ) ) <EOL> dsk = { } <EOL> for i in range ( df . npartitions ) : <EOL> dsk [ name , i ] = ( _rename , metadata , ( df . _name , i ) ) <EOL> return _Frame ( merge ( dsk , df . dask ) , name , metadata , df . divisions ) <EOL> def quantile ( df , q ) : <EOL> """<STR_LIT>""" <EOL> assert ( isinstance ( df , DataFrame ) and len ( df . columns ) == <NUM_LIT:1> or <EOL> isinstance ( df , Series ) ) <EOL> from dask . array . percentile import _percentile , merge_percentiles <EOL> if isinstance ( q , ( list , tuple , np . ndarray ) ) : <EOL> merge_type = lambda v : pd . Series ( v , index = q , name = df . name ) <EOL> return_type = df . _constructor <EOL> if issubclass ( return_type , Index ) : <EOL> return_type = Series <EOL> else : <EOL> merge_type = lambda v : df . _partition_type ( v ) . item ( ) <EOL> return_type = df . _constructor_sliced <EOL> q = [ q ] <EOL> qs = np . asarray ( q ) * <NUM_LIT:100> <EOL> token = tokenize ( df , qs ) <EOL> if len ( qs ) == <NUM_LIT:0> : <EOL> name = '<STR_LIT>' + token <EOL> empty_index = pd . Index ( [ ] , dtype = float ) <EOL> return Series ( { ( name , <NUM_LIT:0> ) : pd . Series ( [ ] , name = df . name , index = empty_index ) } , <EOL> name , df . name , [ None , None ] ) <EOL> else : <EOL> new_divisions = [ np . min ( q ) , np . max ( q ) ] <EOL> name = '<STR_LIT>' + token <EOL> val_dsk = dict ( ( ( name , i ) , ( _percentile , ( getattr , key , '<STR_LIT>' ) , qs ) ) <EOL> for i , key in enumerate ( df . _keys ( ) ) ) <EOL> name2 = '<STR_LIT>' + token <EOL> len_dsk = dict ( ( ( name2 , i ) , ( len , key ) ) for i , key in enumerate ( df . _keys ( ) ) ) <EOL> name3 = '<STR_LIT>' + token <EOL> merge_dsk = { ( name3 , <NUM_LIT:0> ) : ( merge_type , ( merge_percentiles , qs , [ qs ] * df . npartitions , <EOL> sorted ( val_dsk ) , sorted ( len_dsk ) ) ) } <EOL> dsk = merge ( df . dask , val_dsk , len_dsk , merge_dsk ) <EOL> return return_type ( dsk , name3 , df . name , new_divisions ) <EOL> def cov_corr ( df , min_periods = None , corr = False , scalar = False ) : <EOL> """<STR_LIT>""" <EOL> if min_periods is None : <EOL> min_periods = <NUM_LIT:2> <EOL> elif min_periods < <NUM_LIT:2> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> prefix = '<STR_LIT>' if corr else '<STR_LIT>' <EOL> df = df . _get_numeric_data ( ) <EOL> name = '<STR_LIT>' . format ( prefix , tokenize ( df , min_periods , scalar ) ) <EOL> if scalar and len ( df . columns ) != <NUM_LIT:2> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> k = '<STR_LIT>' . format ( prefix , df . _name ) <EOL> dsk = dict ( ( ( k , i ) , ( cov_corr_chunk , f , corr ) ) <EOL> for ( i , f ) in enumerate ( df . _keys ( ) ) ) <EOL> dsk [ ( name , <NUM_LIT:0> ) ] = ( cov_corr_agg , list ( dsk . keys ( ) ) , df . _pd , min_periods , <EOL> corr , scalar ) <EOL> dsk = merge ( df . dask , dsk ) <EOL> if scalar : <EOL> return Scalar ( dsk , name ) <EOL> return DataFrame ( dsk , name , df . _pd , ( df . columns [ <NUM_LIT:0> ] , df . columns [ - <NUM_LIT:1> ] ) ) <EOL> def cov_corr_chunk ( df , corr = False ) : <EOL> """<STR_LIT>""" <EOL> mat = df . values <EOL> mask = np . isfinite ( mat ) <EOL> keep = np . bitwise_and ( mask [ : , None , : ] , mask [ : , : , None ] ) <EOL> x = np . where ( keep , mat [ : , None , : ] , np . nan ) <EOL> sums = np . nansum ( x , <NUM_LIT:0> ) <EOL> counts = keep . astype ( '<STR_LIT:int>' ) . sum ( <NUM_LIT:0> ) <EOL> cov = df . cov ( ) . values <EOL> dtype = [ ( '<STR_LIT>' , sums . dtype ) , ( '<STR_LIT:count>' , counts . dtype ) , ( '<STR_LIT>' , cov . dtype ) ] <EOL> if corr : <EOL> m = np . nansum ( ( x - sums / np . where ( counts , counts , np . nan ) ) ** <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> dtype . append ( ( '<STR_LIT:m>' , m . dtype ) ) <EOL> out = np . empty ( counts . shape , dtype = dtype ) <EOL> out [ '<STR_LIT>' ] = sums <EOL> out [ '<STR_LIT:count>' ] = counts <EOL> out [ '<STR_LIT>' ] = cov * ( counts - <NUM_LIT:1> ) <EOL> if corr : <EOL> out [ '<STR_LIT:m>' ] = m <EOL> return out <EOL> def cov_corr_agg ( data , meta , min_periods = <NUM_LIT:2> , corr = False , scalar = False ) : <EOL> """<STR_LIT>""" <EOL> data = np . concatenate ( data ) . reshape ( ( len ( data ) , ) + data [ <NUM_LIT:0> ] . shape ) <EOL> sums = np . nan_to_num ( data [ '<STR_LIT>' ] ) <EOL> counts = data [ '<STR_LIT:count>' ] <EOL> cum_sums = np . cumsum ( sums , <NUM_LIT:0> ) <EOL> cum_counts = np . cumsum ( counts , <NUM_LIT:0> ) <EOL> s1 = cum_sums [ : - <NUM_LIT:1> ] <EOL> s2 = sums [ <NUM_LIT:1> : ] <EOL> n1 = cum_counts [ : - <NUM_LIT:1> ] <EOL> n2 = counts [ <NUM_LIT:1> : ] <EOL> d = ( s2 / n2 ) - ( s1 / n1 ) <EOL> C = ( np . nansum ( ( n1 * n2 ) / ( n1 + n2 ) * ( d * d . transpose ( ( <NUM_LIT:0> , <NUM_LIT:2> , <NUM_LIT:1> ) ) ) , <NUM_LIT:0> ) + <EOL> np . nansum ( data [ '<STR_LIT>' ] , <NUM_LIT:0> ) ) <EOL> C [ cum_counts [ - <NUM_LIT:1> ] < min_periods ] = np . nan <EOL> nobs = np . where ( cum_counts [ - <NUM_LIT:1> ] , cum_counts [ - <NUM_LIT:1> ] , np . nan ) <EOL> if corr : <EOL> mu = cum_sums [ - <NUM_LIT:1> ] / nobs <EOL> counts_na = np . where ( counts , counts , np . nan ) <EOL> m2 = np . nansum ( data [ '<STR_LIT:m>' ] + counts * ( sums / counts_na - mu ) ** <NUM_LIT:2> , axis = <NUM_LIT:0> ) <EOL> den = np . sqrt ( m2 * m2 . T ) <EOL> else : <EOL> den = nobs - <NUM_LIT:1> <EOL> mat = C / den <EOL> if scalar : <EOL> return mat [ <NUM_LIT:0> , <NUM_LIT:1> ] <EOL> return pd . DataFrame ( mat , columns = meta . columns , index = meta . columns ) <EOL> def pd_split ( df , p , random_state = None ) : <EOL> """<STR_LIT>""" <EOL> p = list ( p ) <EOL> index = pseudorandom ( len ( df ) , p , random_state ) <EOL> return [ df . iloc [ index == i ] for i in range ( len ( p ) ) ] <EOL> def _take_last ( a , skipna = True ) : <EOL> """<STR_LIT>""" <EOL> if skipna is False : <EOL> return a . iloc [ - <NUM_LIT:1> ] <EOL> else : <EOL> group_dummy = np . ones ( len ( a . index ) ) <EOL> last_row = a . groupby ( group_dummy ) . last ( ) <EOL> if isinstance ( a , pd . DataFrame ) : <EOL> return pd . Series ( last_row . values [ <NUM_LIT:0> ] , index = a . columns ) <EOL> else : <EOL> return last_row . values [ <NUM_LIT:0> ] <EOL> def repartition_divisions ( a , b , name , out1 , out2 , force = False ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( b , ( list , tuple ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> b = list ( b ) <EOL> if len ( b ) < <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if b != sorted ( b ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if len ( b [ : - <NUM_LIT:1> ] ) != len ( list ( unique ( b [ : - <NUM_LIT:1> ] ) ) ) : <EOL> msg = '<STR_LIT>' <EOL> raise ValueError ( msg ) <EOL> if force : <EOL> if a [ <NUM_LIT:0> ] < b [ <NUM_LIT:0> ] : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise ValueError ( msg ) <EOL> if a [ - <NUM_LIT:1> ] > b [ - <NUM_LIT:1> ] : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise ValueError ( msg ) <EOL> else : <EOL> if a [ <NUM_LIT:0> ] != b [ <NUM_LIT:0> ] : <EOL> msg = '<STR_LIT>' <EOL> raise ValueError ( msg ) <EOL> if a [ - <NUM_LIT:1> ] != b [ - <NUM_LIT:1> ] : <EOL> msg = '<STR_LIT>' <EOL> raise ValueError ( msg ) <EOL> def _is_single_last_div ( x ) : <EOL> """<STR_LIT>""" <EOL> return len ( x ) >= <NUM_LIT:2> and x [ - <NUM_LIT:1> ] == x [ - <NUM_LIT:2> ] <EOL> c = [ a [ <NUM_LIT:0> ] ] <EOL> d = dict ( ) <EOL> low = a [ <NUM_LIT:0> ] <EOL> i , j = <NUM_LIT:1> , <NUM_LIT:1> <EOL> k = <NUM_LIT:0> <EOL> last_elem = _is_single_last_div ( a ) <EOL> while ( i < len ( a ) and j < len ( b ) ) : <EOL> if a [ i ] < b [ j ] : <EOL> d [ ( out1 , k ) ] = ( _loc , ( name , i - <NUM_LIT:1> ) , low , a [ i ] , False ) <EOL> low = a [ i ] <EOL> i += <NUM_LIT:1> <EOL> elif a [ i ] > b [ j ] : <EOL> d [ ( out1 , k ) ] = ( _loc , ( name , i - <NUM_LIT:1> ) , low , b [ j ] , False ) <EOL> low = b [ j ] <EOL> j += <NUM_LIT:1> <EOL> else : <EOL> d [ ( out1 , k ) ] = ( _loc , ( name , i - <NUM_LIT:1> ) , low , b [ j ] , False ) <EOL> low = b [ j ] <EOL> i += <NUM_LIT:1> <EOL> j += <NUM_LIT:1> <EOL> c . append ( low ) <EOL> k += <NUM_LIT:1> <EOL> if a [ - <NUM_LIT:1> ] < b [ - <NUM_LIT:1> ] or b [ - <NUM_LIT:1> ] == b [ - <NUM_LIT:2> ] : <EOL> for _j in range ( j , len ( b ) ) : <EOL> m = len ( a ) - <NUM_LIT:2> <EOL> d [ ( out1 , k ) ] = ( _loc , ( name , m ) , low , b [ _j ] , False ) <EOL> low = b [ _j ] <EOL> c . append ( low ) <EOL> k += <NUM_LIT:1> <EOL> else : <EOL> if last_elem and i < len ( a ) : <EOL> d [ ( out1 , k ) ] = ( _loc , ( name , i - <NUM_LIT:1> ) , a [ i ] , a [ i ] , False ) <EOL> k += <NUM_LIT:1> <EOL> c . append ( a [ - <NUM_LIT:1> ] ) <EOL> d [ ( out1 , k - <NUM_LIT:1> ) ] = d [ ( out1 , k - <NUM_LIT:1> ) ] [ : - <NUM_LIT:1> ] + ( True , ) <EOL> i , j = <NUM_LIT:0> , <NUM_LIT:1> <EOL> last_elem = _is_single_last_div ( c ) <EOL> while j < len ( b ) : <EOL> tmp = [ ] <EOL> while c [ i ] < b [ j ] : <EOL> tmp . append ( ( out1 , i ) ) <EOL> i += <NUM_LIT:1> <EOL> if last_elem and c [ i ] == b [ - <NUM_LIT:1> ] and ( b [ - <NUM_LIT:1> ] != b [ - <NUM_LIT:2> ] or j == len ( b ) - <NUM_LIT:1> ) and i < k : <EOL> tmp . append ( ( out1 , i ) ) <EOL> i += <NUM_LIT:1> <EOL> if len ( tmp ) == <NUM_LIT:0> : <EOL> d [ ( out2 , j - <NUM_LIT:1> ) ] = ( _loc , ( name , <NUM_LIT:0> ) , a [ <NUM_LIT:0> ] , a [ <NUM_LIT:0> ] , False ) <EOL> elif len ( tmp ) == <NUM_LIT:1> : <EOL> d [ ( out2 , j - <NUM_LIT:1> ) ] = tmp [ <NUM_LIT:0> ] <EOL> else : <EOL> if not tmp : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( pformat ( a ) , pformat ( b ) , pformat ( c ) ) ) <EOL> d [ ( out2 , j - <NUM_LIT:1> ) ] = ( pd . concat , ( list , tmp ) ) <EOL> j += <NUM_LIT:1> <EOL> return d <EOL> def repartition_npartitions ( df , npartitions ) : <EOL> """<STR_LIT>""" <EOL> npartitions = min ( npartitions , df . npartitions ) <EOL> k = int ( math . ceil ( df . npartitions / npartitions ) ) <EOL> divisions = df . divisions [ : : k ] <EOL> if len ( divisions ) <= npartitions : <EOL> divisions = divisions + ( df . divisions [ - <NUM_LIT:1> ] , ) <EOL> return df . repartition ( divisions = divisions ) <EOL> def repartition ( df , divisions = None , force = False ) : <EOL> """<STR_LIT>""" <EOL> token = tokenize ( df , divisions ) <EOL> if isinstance ( df , _Frame ) : <EOL> tmp = '<STR_LIT>' + token <EOL> out = '<STR_LIT>' + token <EOL> dsk = repartition_divisions ( df . divisions , divisions , <EOL> df . _name , tmp , out , force = force ) <EOL> return df . _constructor ( merge ( df . dask , dsk ) , out , <EOL> df . _pd , divisions ) <EOL> elif isinstance ( df , ( pd . Series , pd . DataFrame ) ) : <EOL> name = '<STR_LIT>' + token <EOL> from . utils import shard_df_on_index <EOL> dfs = shard_df_on_index ( df , divisions [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) <EOL> dsk = dict ( ( ( name , i ) , df ) for i , df in enumerate ( dfs ) ) <EOL> return _Frame ( dsk , name , df , divisions ) <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> class Accessor ( object ) : <EOL> def __init__ ( self , series ) : <EOL> if not isinstance ( series , Series ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . _series = series <EOL> def _property_map ( self , key ) : <EOL> return map_partitions ( self . getattr , self . _series . name , self . _series , key ) <EOL> def _function_map ( self , key , * args ) : <EOL> return map_partitions ( self . call , self . _series . name , self . _series , key , <EOL> * args ) <EOL> def __dir__ ( self ) : <EOL> return sorted ( set ( dir ( type ( self ) ) + list ( self . __dict__ ) + <EOL> dir ( self . ns ) ) ) <EOL> def __getattr__ ( self , key ) : <EOL> if key in dir ( self . ns ) : <EOL> if isinstance ( getattr ( self . ns , key ) , property ) : <EOL> return self . _property_map ( key ) <EOL> else : <EOL> return partial ( self . _function_map , key ) <EOL> else : <EOL> raise AttributeError ( key ) <EOL> class DatetimeAccessor ( Accessor ) : <EOL> """<STR_LIT>""" <EOL> ns = pd . Series . dt <EOL> @ staticmethod <EOL> def getattr ( obj , attr ) : <EOL> return getattr ( obj . dt , attr ) <EOL> @ staticmethod <EOL> def call ( obj , attr , * args ) : <EOL> return getattr ( obj . dt , attr ) ( * args ) <EOL> class StringAccessor ( Accessor ) : <EOL> """<STR_LIT>""" <EOL> ns = pd . Series . str <EOL> @ staticmethod <EOL> def getattr ( obj , attr ) : <EOL> return getattr ( obj . str , attr ) <EOL> @ staticmethod <EOL> def call ( obj , attr , * args ) : <EOL> return getattr ( obj . str , attr ) ( * args ) <EOL> def try_loc ( df , ind ) : <EOL> try : <EOL> return df . loc [ ind ] <EOL> except KeyError : <EOL> return df . head ( <NUM_LIT:0> ) </s>
