<s> import pytest <EOL> import numpy as np <EOL> import pandas as pd <EOL> import pandas . util . testing as tm <EOL> dd = pytest . importorskip ( '<STR_LIT>' ) <EOL> from blaze import symbol , discover , compute , broadcast_collect <EOL> from blaze . expr import mean , count , sum , min , max , var , std , summary <EOL> def eq ( a , b ) : <EOL> if isinstance ( a , dd . DataFrame ) : <EOL> tm . assert_frame_equal ( a . compute ( ) , b ) <EOL> elif isinstance ( b , dd . Series ) : <EOL> tm . assert_series_equal ( a . compute ( ) , b ) <EOL> t = symbol ( '<STR_LIT:t>' , '<STR_LIT>' ) <EOL> nt = symbol ( '<STR_LIT:t>' , '<STR_LIT>' ) <EOL> df = pd . DataFrame ( [ [ '<STR_LIT>' , <NUM_LIT:100> , <NUM_LIT:1> ] , <EOL> [ '<STR_LIT>' , <NUM_LIT:200> , <NUM_LIT:2> ] , <EOL> [ '<STR_LIT>' , <NUM_LIT:50> , <NUM_LIT:3> ] ] , columns = [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT:id>' ] ) <EOL> ndf = pd . DataFrame ( [ [ '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> ] , <EOL> [ '<STR_LIT>' , np . nan , <NUM_LIT:2> ] , <EOL> [ np . nan , <NUM_LIT> , <NUM_LIT:3> ] ] , columns = [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT:id>' ] ) <EOL> tbig = symbol ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> dfbig = pd . DataFrame ( [ [ '<STR_LIT>' , '<STR_LIT:F>' , <NUM_LIT:100> , <NUM_LIT:1> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:F>' , <NUM_LIT:100> , <NUM_LIT:3> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:F>' , <NUM_LIT:100> , <NUM_LIT:4> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:M>' , <NUM_LIT:100> , <NUM_LIT:5> ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:M>' , <NUM_LIT:200> , <NUM_LIT:5> ] ] , <EOL> columns = [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:id>' ] ) <EOL> ddf = dd . from_pandas ( df , npartitions = <NUM_LIT:2> ) <EOL> ddfbig = dd . from_pandas ( dfbig , npartitions = <NUM_LIT:2> ) <EOL> def test_broadcast ( ) : <EOL> bcast = broadcast_collect ( expr = t . amount * t . id ) <EOL> result = compute ( bcast , ddf ) <EOL> eq ( result , df . amount * df . id ) <EOL> def test_symbol ( ) : <EOL> eq ( compute ( t , ddf ) , df ) <EOL> def test_sample ( ) : <EOL> assert len ( compute ( t . sample ( n = <NUM_LIT:2> ) , ddf ) ) == len ( df . sample ( n = <NUM_LIT:2> ) ) <EOL> assert len ( compute ( t . sample ( frac = <NUM_LIT> ) , ddf ) ) == len ( df . sample ( frac = <NUM_LIT> ) ) <EOL> def test_head ( ) : <EOL> eq ( compute ( t . head ( <NUM_LIT:1> ) , ddf ) , df . head ( <NUM_LIT:1> ) ) <EOL> def test_tail ( ) : <EOL> eq ( compute ( t . tail ( <NUM_LIT:1> ) , ddf ) , df . tail ( <NUM_LIT:1> ) ) <EOL> def test_relabel ( ) : <EOL> result = compute ( t . relabel ( { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:id>' : '<STR_LIT>' } ) , ddf ) <EOL> expected = df . rename ( columns = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT:id>' : '<STR_LIT>' } ) <EOL> eq ( result , expected ) <EOL> def test_relabel_series ( ) : <EOL> result = compute ( t . relabel ( { '<STR_LIT:name>' : '<STR_LIT>' } ) , ddf . name ) <EOL> assert result . name == '<STR_LIT>' <EOL> def test_series_columnwise ( ) : <EOL> data = pd . Series ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] , name = '<STR_LIT:a>' ) <EOL> s = dd . from_pandas ( data , npartitions = <NUM_LIT:2> ) <EOL> t = symbol ( '<STR_LIT:t>' , '<STR_LIT>' ) <EOL> result = compute ( t . a + <NUM_LIT:1> , s ) <EOL> eq ( result , data ) <EOL> def test_projection ( ) : <EOL> eq ( compute ( t [ [ '<STR_LIT:name>' , '<STR_LIT:id>' ] ] , ddf ) , df [ [ '<STR_LIT:name>' , '<STR_LIT:id>' ] ] ) <EOL> def test_field_on_series ( ) : <EOL> expr = symbol ( '<STR_LIT:s>' , '<STR_LIT>' ) <EOL> data = pd . Series ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] , name = '<STR_LIT:s>' ) <EOL> s = dd . from_pandas ( data , npartitions = <NUM_LIT:2> ) <EOL> eq ( compute ( expr . s , s ) , data ) <EOL> def test_selection ( ) : <EOL> eq ( compute ( t [ t [ '<STR_LIT>' ] == <NUM_LIT:0> ] , ddf ) , df [ df [ '<STR_LIT>' ] == <NUM_LIT:0> ] ) <EOL> eq ( compute ( t [ t [ '<STR_LIT>' ] > <NUM_LIT> ] , ddf ) , df [ df [ '<STR_LIT>' ] > <NUM_LIT> ] ) <EOL> def test_selection_out_of_order ( ) : <EOL> expr = t [ '<STR_LIT:name>' ] [ t [ '<STR_LIT>' ] < <NUM_LIT:100> ] <EOL> expected = df . loc [ df . amount < <NUM_LIT:100> , '<STR_LIT:name>' ] <EOL> result = compute ( expr , ddf ) <EOL> eq ( result , expected ) <EOL> def test_selection_inner_inputs ( ) : <EOL> s_data = pd . DataFrame ( { '<STR_LIT:a>' : np . arange ( <NUM_LIT:5> ) } ) <EOL> t_data = pd . DataFrame ( { '<STR_LIT:a>' : np . arange ( <NUM_LIT:5> ) } ) <EOL> s_dd = dd . from_pandas ( s_data , npartitions = <NUM_LIT:2> ) <EOL> t_dd = dd . from_pandas ( t_data , npartitions = <NUM_LIT:2> ) <EOL> s = symbol ( '<STR_LIT:s>' , '<STR_LIT>' ) <EOL> t = symbol ( '<STR_LIT:t>' , '<STR_LIT>' ) <EOL> eq ( compute ( s [ s . a == t . a ] , { s : s_dd , t : t_dd } ) , s_data ) <EOL> def test_reductions ( ) : <EOL> assert compute ( mean ( t [ '<STR_LIT>' ] ) , ddf ) == <NUM_LIT> / <NUM_LIT:3> <EOL> assert compute ( count ( t [ '<STR_LIT>' ] ) , ddf ) == <NUM_LIT:3> <EOL> assert compute ( sum ( t [ '<STR_LIT>' ] ) , ddf ) == <NUM_LIT:100> + <NUM_LIT:200> + <NUM_LIT:50> <EOL> assert compute ( min ( t [ '<STR_LIT>' ] ) , ddf ) == <NUM_LIT:50> <EOL> assert compute ( max ( t [ '<STR_LIT>' ] ) , ddf ) == <NUM_LIT:200> <EOL> assert compute ( var ( t [ '<STR_LIT>' ] ) , ddf ) == df . amount . var ( ddof = <NUM_LIT:0> ) <EOL> assert compute ( var ( t [ '<STR_LIT>' ] , unbiased = True ) , ddf ) == df . amount . var ( ) <EOL> assert compute ( std ( t [ '<STR_LIT>' ] ) , ddf ) == df . amount . std ( ddof = <NUM_LIT:0> ) <EOL> assert compute ( std ( t [ '<STR_LIT>' ] , unbiased = True ) , ddf ) == df . amount . std ( ) <EOL> def test_summary ( ) : <EOL> expr = summary ( count = t . id . count ( ) , sum = t . amount . sum ( ) ) <EOL> eq ( compute ( expr , ddf ) , pd . Series ( { '<STR_LIT:count>' : <NUM_LIT:3> , '<STR_LIT>' : <NUM_LIT> } ) ) <EOL> def test_summary_on_series ( ) : <EOL> ser = dd . from_pandas ( pd . Series ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) , npartitions = <NUM_LIT:2> ) <EOL> s = symbol ( '<STR_LIT:s>' , '<STR_LIT>' ) <EOL> expr = summary ( max = s . max ( ) , min = s . min ( ) ) <EOL> assert compute ( expr , ser ) == ( <NUM_LIT:3> , <NUM_LIT:1> ) <EOL> expr = summary ( max = s . max ( ) , min = s . min ( ) , keepdims = True ) <EOL> assert compute ( expr , ser ) == [ ( <NUM_LIT:3> , <NUM_LIT:1> ) ] <EOL> @ pytest . mark . parametrize ( '<STR_LIT>' , [ [ <NUM_LIT:1> ] , [ <NUM_LIT:2> , <NUM_LIT:3> ] ] ) <EOL> def test_isin ( keys ) : <EOL> expr = t [ t . id . isin ( keys ) ] <EOL> result = compute ( expr , ddf ) <EOL> expected = df . loc [ df . id . isin ( keys ) ] <EOL> eq ( result , expected ) <EOL> def test_coerce_series ( ) : <EOL> s = pd . Series ( list ( '<STR_LIT>' ) , name = '<STR_LIT:a>' ) <EOL> dds = dd . from_pandas ( s , npartitions = <NUM_LIT:2> ) <EOL> t = symbol ( '<STR_LIT:t>' , discover ( s ) ) <EOL> result = compute ( t . coerce ( to = '<STR_LIT>' ) , dds ) <EOL> expected = pd . Series ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] , name = s . name ) <EOL> eq ( result , expected ) <EOL> def test_nelements_count ( ) : <EOL> assert compute ( t . nelements ( ) , ddf ) == len ( df ) <EOL> assert compute ( t . nrows , ddf ) == len ( df ) <EOL> assert compute ( count ( t ) , ddf ) == len ( df ) </s>
