<s> from __future__ import absolute_import <EOL> from . base import Node , Stack <EOL> from . . dq . field_statistics import FieldStatistics <EOL> from . . metadata import FieldMap , FieldList , Field <EOL> import logging <EOL> import itertools <EOL> import random <EOL> class SampleNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT:size>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , size = <NUM_LIT:1000> , discard_sample = False , method = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> super ( SampleNode , self ) . __init__ ( ) <EOL> self . size = size <EOL> self . discard_sample = discard_sample <EOL> self . method = method <EOL> if method == "<STR_LIT>" : <EOL> self . stack = Stack ( size ) <EOL> else : <EOL> self . stack = None <EOL> if method == "<STR_LIT>" and ( ( size > <NUM_LIT:100> ) or ( size < <NUM_LIT:0> ) ) : <EOL> raise ValueError , "<STR_LIT>" <EOL> def run ( self ) : <EOL> pipe = self . input <EOL> count = <NUM_LIT:0> <EOL> for row in pipe . rows ( ) : <EOL> logging . debug ( "<STR_LIT>" % count ) <EOL> if self . method == "<STR_LIT>" : <EOL> uniform = random . random ( ) <EOL> self . stack . push ( key = uniform , value = row ) <EOL> elif self . method == "<STR_LIT>" : <EOL> if random . random ( ) < float ( self . size ) / <NUM_LIT> : <EOL> self . put ( row ) <EOL> count += <NUM_LIT:1> <EOL> else : <EOL> self . put ( row ) <EOL> count += <NUM_LIT:1> <EOL> if count >= self . size : <EOL> break <EOL> if self . stack : <EOL> for row in self . stack . items ( ) : <EOL> self . put ( row ) <EOL> class AppendNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( AppendNode , self ) . __init__ ( ) <EOL> @ property <EOL> def output_fields ( self ) : <EOL> if not self . inputs : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return self . inputs [ <NUM_LIT:0> ] . fields <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> for pipe in self . inputs : <EOL> for row in pipe . rows ( ) : <EOL> self . put ( row ) <EOL> class MergeNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , joins = None , master = None , maps = None ) : <EOL> super ( MergeNode , self ) . __init__ ( ) <EOL> if joins : <EOL> self . joins = joins <EOL> else : <EOL> self . joins = [ ] <EOL> if master : <EOL> self . master = master <EOL> else : <EOL> self . master = <NUM_LIT:0> <EOL> self . maps = maps <EOL> self . _output_fields = [ ] <EOL> def initialize ( self ) : <EOL> pass <EOL> self . _keys = { } <EOL> self . _kindexes = { } <EOL> self . master_input = self . inputs [ self . master ] <EOL> self . detail_inputs = [ ] <EOL> for ( tag , pipe ) in enumerate ( self . inputs ) : <EOL> if pipe is not self . master_input : <EOL> self . detail_inputs . append ( ( tag , pipe ) ) <EOL> for join in self . joins : <EOL> joinlen = len ( join ) <EOL> if joinlen == <NUM_LIT:3> : <EOL> ( detail_tag , master_key , detail_key ) = join <EOL> elif joinlen == <NUM_LIT:2> : <EOL> ( detail_tag , master_key ) = join <EOL> detail_key = master_key <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if not ( type ( detail_key ) == list or type ( detail_key ) == tuple ) : <EOL> detail_key = ( detail_key , ) <EOL> if not ( type ( master_key ) == list or type ( master_key ) == tuple ) : <EOL> master_key = ( master_key , ) <EOL> if detail_tag == self . master : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> self . _keys [ detail_tag ] = ( detail_key , master_key ) <EOL> detail_input = self . inputs [ detail_tag ] <EOL> detail_indexes = detail_input . fields . indexes ( detail_key ) <EOL> master_indexes = self . master_input . fields . indexes ( master_key ) <EOL> self . _kindexes [ detail_tag ] = ( detail_indexes , master_indexes ) <EOL> self . _input_rows = { } <EOL> for ( tag , pipe ) in enumerate ( self . inputs ) : <EOL> self . _input_rows [ tag ] = { } <EOL> self . _filters = { } <EOL> self . _maps = { } <EOL> if self . maps : <EOL> for ( tag , fmap ) in self . maps . items ( ) : <EOL> if type ( fmap ) == dict : <EOL> fmap = FieldMap ( rename = fmap . get ( "<STR_LIT>" ) , drop = fmap . get ( "<STR_LIT>" ) , keep = fmap . get ( "<STR_LIT>" ) ) <EOL> elif type ( fmap ) != FieldMap : <EOL> raise Exception ( "<STR_LIT>" % type ( fmap ) ) <EOL> f = fmap . row_filter ( self . inputs [ tag ] . fields ) <EOL> self . _maps [ tag ] = fmap <EOL> self . _filters [ tag ] = f <EOL> fields = [ ] <EOL> for ( tag , pipe ) in enumerate ( self . inputs ) : <EOL> fmap = self . _maps . get ( tag , None ) <EOL> if fmap : <EOL> fields += fmap . map ( pipe . fields ) <EOL> else : <EOL> fields += pipe . fields <EOL> self . _output_fields = FieldList ( fields ) <EOL> @ property <EOL> def output_fields ( self ) : <EOL> return self . _output_fields <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> for ( tag , pipe ) in self . detail_inputs : <EOL> detail = self . _input_rows [ tag ] <EOL> key_indexes = self . _kindexes [ tag ] [ <NUM_LIT:0> ] <EOL> self . _read_input ( tag , pipe , key_indexes , detail ) <EOL> rfilter = self . _filters . get ( self . master ) <EOL> for row in self . master_input . rows ( ) : <EOL> if rfilter : <EOL> joined_row = rfilter . filter ( row [ : ] ) <EOL> else : <EOL> joined_row = row [ : ] <EOL> joined = False <EOL> for ( tag , pipe ) in self . detail_inputs : <EOL> detail_data = self . _input_rows [ tag ] <EOL> key = [ ] <EOL> for i in self . _kindexes [ tag ] [ <NUM_LIT:1> ] : <EOL> key . append ( row [ i ] ) <EOL> key = tuple ( key ) <EOL> detail = detail_data . get ( tuple ( key ) ) <EOL> if not detail : <EOL> joined = False <EOL> break <EOL> else : <EOL> joined = True <EOL> joined_row += detail <EOL> if joined : <EOL> self . put ( joined_row ) <EOL> def _read_input ( self , tag , pipe , key_indexes , detail ) : <EOL> rfilter = self . _filters . get ( tag ) <EOL> for row in pipe . rows ( ) : <EOL> key = [ ] <EOL> for i in key_indexes : <EOL> key . append ( row [ i ] ) <EOL> if rfilter : <EOL> detail [ tuple ( key ) ] = rfilter . filter ( row ) <EOL> else : <EOL> detail [ tuple ( key ) ] = row <EOL> class DistinctNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , distinct_fields = None , discard = False ) : <EOL> """<STR_LIT>""" <EOL> super ( DistinctNode , self ) . __init__ ( ) <EOL> if distinct_fields : <EOL> self . distinct_fields = distinct_fields <EOL> else : <EOL> self . distinct_fields = [ ] <EOL> self . discard = discard <EOL> def initialize ( self ) : <EOL> field_map = FieldMap ( keep = self . distinct_fields ) <EOL> self . row_filter = field_map . row_filter ( self . input_fields ) <EOL> def run ( self ) : <EOL> pipe = self . input <EOL> self . distinct_values = set ( ) <EOL> if not self . distinct_fields : <EOL> for row in pipe . rows ( ) : <EOL> self . put ( row ) <EOL> return <EOL> for row in pipe . rows ( ) : <EOL> key_tuple = tuple ( self . row_filter ( row ) ) <EOL> if key_tuple not in self . distinct_values : <EOL> self . distinct_values . add ( key_tuple ) <EOL> if not self . discard : <EOL> self . put ( row ) <EOL> else : <EOL> if self . discard : <EOL> self . put ( row ) <EOL> class Aggregate ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . count = <NUM_LIT:0> <EOL> self . sum = <NUM_LIT:0> <EOL> self . min = <NUM_LIT:0> <EOL> self . max = <NUM_LIT:0> <EOL> self . average = None <EOL> def aggregate_value ( self , value ) : <EOL> self . count += <NUM_LIT:1> <EOL> self . sum += value <EOL> self . min = min ( self . min , value ) <EOL> self . max = max ( self . max , value ) <EOL> def finalize ( self ) : <EOL> if self . count : <EOL> self . average = self . sum / self . count <EOL> else : <EOL> self . average = None <EOL> class KeyAggregate ( object ) : <EOL> def __init__ ( self ) : <EOL> self . count = <NUM_LIT:0> <EOL> self . field_aggregates = { } <EOL> class AggregateNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , keys = None , measures = None , default_aggregations = None , <EOL> record_count_field = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> super ( AggregateNode , self ) . __init__ ( ) <EOL> if default_aggregations is None : <EOL> default_aggregations = [ "<STR_LIT>" ] <EOL> if keys : <EOL> self . key_fields = keys <EOL> else : <EOL> self . key_fields = [ ] <EOL> self . aggregations = { } <EOL> self . record_count_field = record_count_field <EOL> self . measures = measures or [ ] <EOL> def add_measure ( self , field , aggregations = None ) : <EOL> """<STR_LIT>""" <EOL> self . aggregations [ field ] = aggregations <EOL> self . measures . append ( field ) <EOL> @ property <EOL> def output_fields ( self ) : <EOL> fields = FieldList ( ) <EOL> if self . key_fields : <EOL> for field in self . input_fields . fields ( self . key_fields ) : <EOL> fields . append ( field ) <EOL> for field in self . measures : <EOL> fields . append ( Field ( field + "<STR_LIT>" , storage_type = "<STR_LIT:float>" , analytical_type = "<STR_LIT>" ) ) <EOL> fields . append ( Field ( field + "<STR_LIT>" , storage_type = "<STR_LIT:float>" , analytical_type = "<STR_LIT>" ) ) <EOL> fields . append ( Field ( field + "<STR_LIT>" , storage_type = "<STR_LIT:float>" , analytical_type = "<STR_LIT>" ) ) <EOL> fields . append ( Field ( field + "<STR_LIT>" , storage_type = "<STR_LIT:float>" , analytical_type = "<STR_LIT>" ) ) <EOL> fields . append ( Field ( self . record_count_field , storage_type = "<STR_LIT>" , analytical_type = "<STR_LIT>" ) ) <EOL> return fields <EOL> def run ( self ) : <EOL> pipe = self . input <EOL> self . aggregates = { } <EOL> self . keys = [ ] <EOL> self . counts = { } <EOL> key_selectors = self . input_fields . selectors ( self . key_fields ) <EOL> measure_indexes = self . input_fields . indexes ( self . measures ) <EOL> for row in pipe . rows ( ) : <EOL> key = tuple ( itertools . compress ( row , key_selectors ) ) <EOL> if key not in self . keys : <EOL> self . keys . append ( key ) <EOL> key_aggregate = KeyAggregate ( ) <EOL> self . aggregates [ key ] = key_aggregate <EOL> else : <EOL> key_aggregate = self . aggregates [ key ] <EOL> key_aggregate . count += <NUM_LIT:1> <EOL> for i in measure_indexes : <EOL> if i not in key_aggregate . field_aggregates : <EOL> aggregate = Aggregate ( ) <EOL> key_aggregate . field_aggregates [ i ] = aggregate <EOL> else : <EOL> aggregate = key_aggregate . field_aggregates [ i ] <EOL> value = row [ i ] <EOL> aggregate . aggregate_value ( value ) <EOL> for key in self . keys : <EOL> row = list ( key [ : ] ) <EOL> key_aggregate = self . aggregates [ key ] <EOL> for i in measure_indexes : <EOL> aggregate = key_aggregate . field_aggregates [ i ] <EOL> aggregate . finalize ( ) <EOL> row . append ( aggregate . sum ) <EOL> row . append ( aggregate . min ) <EOL> row . append ( aggregate . max ) <EOL> row . append ( aggregate . average ) <EOL> row . append ( key_aggregate . count ) <EOL> self . put ( row ) <EOL> class SelectNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:False>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , condition = None , discard = False ) : <EOL> """<STR_LIT>""" <EOL> super ( SelectNode , self ) . __init__ ( ) <EOL> self . condition = condition <EOL> self . discard = discard <EOL> def initialize ( self ) : <EOL> if isinstance ( self . condition , basestring ) : <EOL> self . _expression = compile ( self . condition , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . _condition_callable = self . _eval_expression <EOL> else : <EOL> self . _condition_callable = self . condition <EOL> def _eval_expression ( self , ** record ) : <EOL> return eval ( self . _expression , None , record ) <EOL> def run ( self ) : <EOL> for record in self . input . records ( ) : <EOL> if self . _condition_callable ( ** record ) : <EOL> self . put_record ( record ) <EOL> class FunctionSelectNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> ] <EOL> } <EOL> def __init__ ( self , function = None , fields = None , discard = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> super ( FunctionSelectNode , self ) . __init__ ( ) <EOL> self . function = function <EOL> self . fields = fields <EOL> self . discard = discard <EOL> self . kwargs = kwargs <EOL> def initialize ( self ) : <EOL> self . indexes = self . input_fields . indexes ( self . fields ) <EOL> def run ( self ) : <EOL> for row in self . input . rows ( ) : <EOL> values = [ row [ index ] for index in self . indexes ] <EOL> flag = self . function ( * values , ** self . kwargs ) <EOL> if ( flag and not self . discard ) or ( not flag and self . discard ) : <EOL> self . put ( row ) <EOL> class SetSelectNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> } , <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:default>" : "<STR_LIT:True>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , field = None , value_set = None , discard = False ) : <EOL> """<STR_LIT>""" <EOL> super ( SetSelectNode , self ) . __init__ ( ) <EOL> self . field = field <EOL> self . value_set = value_set <EOL> self . discard = discard <EOL> def initialize ( self ) : <EOL> self . field_index = self . input_fields . index ( self . field ) <EOL> def run ( self ) : <EOL> for row in self . input . rows ( ) : <EOL> flag = row [ self . field_index ] in self . value_set <EOL> if ( flag and not self . discard ) or ( not flag and self . discard ) : <EOL> self . put ( row ) <EOL> class AuditNode ( Node ) : <EOL> """<STR_LIT>""" <EOL> node_info = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> } <EOL> ] <EOL> } <EOL> def __init__ ( self , distinct_threshold = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> super ( AuditNode , self ) . __init__ ( ) <EOL> self . distinct_threshold = distinct_threshold <EOL> @ property <EOL> def output_fields ( self ) : <EOL> audit_record_fields = [ <EOL> ( "<STR_LIT>" , "<STR_LIT:string>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:float>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT:float>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> ] <EOL> fields = FieldList ( audit_record_fields ) <EOL> return fields <EOL> def initialize ( self ) : <EOL> self . stats = [ ] <EOL> for field in self . input_fields : <EOL> stat = FieldStatistics ( field . name , distinct_threshold = self . distinct_threshold ) <EOL> self . stats . append ( stat ) <EOL> def run ( self ) : <EOL> for row in self . input . rows ( ) : <EOL> for i , value in enumerate ( row ) : <EOL> self . stats [ i ] . probe ( value ) <EOL> for stat in self . stats : <EOL> stat . finalize ( ) <EOL> if stat . distinct_overflow : <EOL> dist_count = None <EOL> else : <EOL> dist_count = len ( stat . distinct_values ) <EOL> row = [ stat . field , <EOL> stat . record_count , <EOL> stat . null_count , <EOL> stat . null_record_ratio , <EOL> stat . empty_string_count , <EOL> dist_count <EOL> ] <EOL> self . put ( row ) </s>
