<s> """<STR_LIT>""" <EOL> import unittest <EOL> import google . cloud . dataflow as df <EOL> from google . cloud . dataflow . pipeline import Pipeline <EOL> from google . cloud . dataflow . transforms import combiners <EOL> import google . cloud . dataflow . transforms . combiners as combine <EOL> from google . cloud . dataflow . transforms . core import CombineGlobally <EOL> from google . cloud . dataflow . transforms . core import Create <EOL> from google . cloud . dataflow . transforms . core import Map <EOL> from google . cloud . dataflow . transforms . ptransform import PTransform <EOL> from google . cloud . dataflow . transforms . util import assert_that , equal_to <EOL> class CombineTest ( unittest . TestCase ) : <EOL> def test_builtin_combines ( self ) : <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> vals = [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] <EOL> mean = sum ( vals ) / float ( len ( vals ) ) <EOL> size = len ( vals ) <EOL> pcoll = pipeline | Create ( '<STR_LIT:start>' , vals ) <EOL> result_mean = pcoll | combine . Mean . Globally ( '<STR_LIT>' ) <EOL> result_count = pcoll | combine . Count . Globally ( '<STR_LIT:count>' ) <EOL> assert_that ( result_mean , equal_to ( [ mean ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_count , equal_to ( [ size ] ) , label = '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( '<STR_LIT>' , [ ( '<STR_LIT:a>' , x ) for x in vals ] ) <EOL> result_key_mean = pcoll | combine . Mean . PerKey ( '<STR_LIT>' ) <EOL> result_key_count = pcoll | combine . Count . PerKey ( '<STR_LIT>' ) <EOL> assert_that ( result_key_mean , equal_to ( [ ( '<STR_LIT:a>' , mean ) ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_key_count , equal_to ( [ ( '<STR_LIT:a>' , size ) ] ) , label = '<STR_LIT>' ) <EOL> pipeline . run ( ) <EOL> def test_top ( self ) : <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> names = { <NUM_LIT:0> : '<STR_LIT>' , <EOL> <NUM_LIT:1> : '<STR_LIT>' , <EOL> <NUM_LIT:2> : '<STR_LIT>' , <EOL> <NUM_LIT:3> : '<STR_LIT>' , <EOL> <NUM_LIT:5> : '<STR_LIT>' , <EOL> <NUM_LIT:6> : '<STR_LIT>' , <EOL> <NUM_LIT:9> : '<STR_LIT>' } <EOL> pcoll = pipeline | Create ( '<STR_LIT:start>' , [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] ) <EOL> result_top = pcoll | combine . Top . Largest ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> result_bot = pcoll | combine . Top . Smallest ( '<STR_LIT>' , <NUM_LIT:4> ) <EOL> result_cmp = pcoll | combine . Top . Of ( <EOL> '<STR_LIT>' , <EOL> <NUM_LIT:6> , <EOL> lambda a , b , names : len ( names [ a ] ) < len ( names [ b ] ) , <EOL> names ) <EOL> assert_that ( result_top , equal_to ( [ [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> ] ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_bot , equal_to ( [ [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_cmp , equal_to ( [ [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> , <NUM_LIT:2> ] ] ) , label = '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( <EOL> '<STR_LIT>' , [ ( '<STR_LIT:a>' , x ) for x in [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] ] ) <EOL> result_key_top = pcoll | combine . Top . LargestPerKey ( '<STR_LIT>' , <NUM_LIT:5> ) <EOL> result_key_bot = pcoll | combine . Top . SmallestPerKey ( '<STR_LIT>' , <NUM_LIT:4> ) <EOL> result_key_cmp = pcoll | combine . Top . PerKey ( <EOL> '<STR_LIT>' , <EOL> <NUM_LIT:6> , <EOL> lambda a , b , names : len ( names [ a ] ) < len ( names [ b ] ) , <EOL> names ) <EOL> assert_that ( result_key_top , equal_to ( [ ( '<STR_LIT:a>' , [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> ] ) ] ) , <EOL> label = '<STR_LIT>' ) <EOL> assert_that ( result_key_bot , equal_to ( [ ( '<STR_LIT:a>' , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) ] ) , <EOL> label = '<STR_LIT>' ) <EOL> assert_that ( result_key_cmp , equal_to ( [ ( '<STR_LIT:a>' , [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> , <NUM_LIT:2> ] ) ] ) , <EOL> label = '<STR_LIT>' ) <EOL> pipeline . run ( ) <EOL> def test_top_shorthands ( self ) : <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( '<STR_LIT:start>' , [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] ) <EOL> result_top = pcoll | df . CombineGlobally ( '<STR_LIT>' , combiners . Largest ( <NUM_LIT:5> ) ) <EOL> result_bot = pcoll | df . CombineGlobally ( '<STR_LIT>' , combiners . Smallest ( <NUM_LIT:4> ) ) <EOL> assert_that ( result_top , equal_to ( [ [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> ] ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_bot , equal_to ( [ [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ] ) , label = '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( <EOL> '<STR_LIT>' , [ ( '<STR_LIT:a>' , x ) for x in [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] ] ) <EOL> result_ktop = pcoll | df . CombinePerKey ( '<STR_LIT>' , combiners . Largest ( <NUM_LIT:5> ) ) <EOL> result_kbot = pcoll | df . CombinePerKey ( '<STR_LIT>' , combiners . Smallest ( <NUM_LIT:4> ) ) <EOL> assert_that ( result_ktop , equal_to ( [ ( '<STR_LIT:a>' , [ <NUM_LIT:9> , <NUM_LIT:6> , <NUM_LIT:6> , <NUM_LIT:5> , <NUM_LIT:3> ] ) ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result_kbot , equal_to ( [ ( '<STR_LIT:a>' , [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ] ) ] ) , label = '<STR_LIT>' ) <EOL> pipeline . run ( ) <EOL> def test_sample ( self ) : <EOL> for ix in xrange ( <NUM_LIT> ) : <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( '<STR_LIT:start>' , [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> result = pcoll | combine . Sample . FixedSizeGlobally ( '<STR_LIT>' % ix , <NUM_LIT:3> ) <EOL> def matcher ( ) : <EOL> def match ( actual ) : <EOL> equal_to ( [ <NUM_LIT:1> ] ) ( [ len ( actual ) ] ) <EOL> equal_to ( [ <NUM_LIT:3> ] ) ( [ len ( actual [ <NUM_LIT:0> ] ) ] ) <EOL> num_ones = sum ( <NUM_LIT:1> for x in actual [ <NUM_LIT:0> ] if x == <NUM_LIT:1> ) <EOL> num_twos = sum ( <NUM_LIT:1> for x in actual [ <NUM_LIT:0> ] if x == <NUM_LIT:2> ) <EOL> equal_to ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ( [ num_ones , num_twos ] ) <EOL> return match <EOL> assert_that ( result , matcher ( ) ) <EOL> pipeline . run ( ) <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> pcoll = pipeline | Create ( <EOL> '<STR_LIT>' , <EOL> sum ( ( [ ( i , <NUM_LIT:1> ) , ( i , <NUM_LIT:1> ) , ( i , <NUM_LIT:2> ) , ( i , <NUM_LIT:2> ) ] for i in xrange ( <NUM_LIT> ) ) , [ ] ) ) <EOL> result = pcoll | combine . Sample . FixedSizePerKey ( '<STR_LIT>' , <NUM_LIT:3> ) <EOL> def matcher ( ) : <EOL> def match ( actual ) : <EOL> for _ , samples in actual : <EOL> equal_to ( [ <NUM_LIT:3> ] ) ( [ len ( samples ) ] ) <EOL> num_ones = sum ( <NUM_LIT:1> for x in samples if x == <NUM_LIT:1> ) <EOL> num_twos = sum ( <NUM_LIT:1> for x in samples if x == <NUM_LIT:2> ) <EOL> equal_to ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ( [ num_ones , num_twos ] ) <EOL> return match <EOL> assert_that ( result , matcher ( ) ) <EOL> pipeline . run ( ) <EOL> def test_tuple_combine_fn ( self ) : <EOL> p = Pipeline ( '<STR_LIT>' ) <EOL> result = ( <EOL> p <EOL> | Create ( [ ( '<STR_LIT:a>' , <NUM_LIT:100> , <NUM_LIT:0.0> ) , ( '<STR_LIT:b>' , <NUM_LIT:10> , - <NUM_LIT:1> ) , ( '<STR_LIT:c>' , <NUM_LIT:1> , <NUM_LIT:100> ) ] ) <EOL> | df . CombineGlobally ( combine . TupleCombineFn ( max , <EOL> combine . MeanCombineFn ( ) , <EOL> sum ) ) . without_defaults ( ) ) <EOL> assert_that ( result , equal_to ( [ ( '<STR_LIT:c>' , <NUM_LIT> / <NUM_LIT:3> , <NUM_LIT> ) ] ) ) <EOL> p . run ( ) <EOL> def test_tuple_combine_fn_without_defaults ( self ) : <EOL> p = Pipeline ( '<STR_LIT>' ) <EOL> result = ( <EOL> p <EOL> | Create ( [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> | df . CombineGlobally ( <EOL> combine . TupleCombineFn ( min , combine . MeanCombineFn ( ) , max ) <EOL> . with_common_input ( ) ) . without_defaults ( ) ) <EOL> assert_that ( result , equal_to ( [ ( <NUM_LIT:1> , <NUM_LIT> / <NUM_LIT:4> , <NUM_LIT:3> ) ] ) ) <EOL> p . run ( ) <EOL> def test_to_list_and_to_dict ( self ) : <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> the_list = [ <NUM_LIT:6> , <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> , <NUM_LIT:0> , <NUM_LIT:6> ] <EOL> pcoll = pipeline | Create ( '<STR_LIT:start>' , the_list ) <EOL> result = pcoll | combine . ToList ( '<STR_LIT>' ) <EOL> def matcher ( expected ) : <EOL> def match ( actual ) : <EOL> equal_to ( expected [ <NUM_LIT:0> ] ) ( actual [ <NUM_LIT:0> ] ) <EOL> return match <EOL> assert_that ( result , matcher ( [ the_list ] ) ) <EOL> pipeline . run ( ) <EOL> pipeline = Pipeline ( '<STR_LIT>' ) <EOL> pairs = [ ( <NUM_LIT:1> , <NUM_LIT:2> ) , ( <NUM_LIT:3> , <NUM_LIT:4> ) , ( <NUM_LIT:5> , <NUM_LIT:6> ) ] <EOL> pcoll = pipeline | Create ( '<STR_LIT>' , pairs ) <EOL> result = pcoll | combine . ToDict ( '<STR_LIT>' ) <EOL> def matcher ( ) : <EOL> def match ( actual ) : <EOL> equal_to ( [ <NUM_LIT:1> ] ) ( [ len ( actual ) ] ) <EOL> equal_to ( pairs ) ( actual [ <NUM_LIT:0> ] . iteritems ( ) ) <EOL> return match <EOL> assert_that ( result , matcher ( ) ) <EOL> pipeline . run ( ) <EOL> def test_combine_globally_with_default ( self ) : <EOL> p = Pipeline ( '<STR_LIT>' ) <EOL> assert_that ( p | Create ( [ ] ) | CombineGlobally ( sum ) , equal_to ( [ <NUM_LIT:0> ] ) ) <EOL> p . run ( ) <EOL> def test_combine_globally_without_default ( self ) : <EOL> p = Pipeline ( '<STR_LIT>' ) <EOL> result = p | Create ( [ ] ) | CombineGlobally ( sum ) . without_defaults ( ) <EOL> assert_that ( result , equal_to ( [ ] ) ) <EOL> p . run ( ) <EOL> def test_combine_globally_with_default_side_input ( self ) : <EOL> class CombineWithSideInput ( PTransform ) : <EOL> def apply ( self , pcoll ) : <EOL> side = pcoll | CombineGlobally ( sum ) . as_singleton_view ( ) <EOL> main = pcoll . pipeline | Create ( [ None ] ) <EOL> return main | Map ( lambda _ , s : s , side ) <EOL> p = Pipeline ( '<STR_LIT>' ) <EOL> result1 = p | Create ( '<STR_LIT>' , [ ] ) | CombineWithSideInput ( '<STR_LIT>' ) <EOL> result2 = p | Create ( '<STR_LIT>' , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] ) | CombineWithSideInput ( '<STR_LIT>' ) <EOL> assert_that ( result1 , equal_to ( [ <NUM_LIT:0> ] ) , label = '<STR_LIT>' ) <EOL> assert_that ( result2 , equal_to ( [ <NUM_LIT:10> ] ) , label = '<STR_LIT>' ) <EOL> p . run ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
