<s> from _functools import partial <EOL> from datetime import datetime , timedelta <EOL> from decimal import Decimal <EOL> import string <EOL> import sys <EOL> import time <EOL> import unittest <EOL> import uuid <EOL> from cassandra import ConsistencyLevel <EOL> from cassandra . cluster import Cluster <EOL> from cassandra . util import uuid_from_time <EOL> from pyspark import SparkConf <EOL> from pyspark . accumulators import AddingAccumulatorParam <EOL> from pyspark . streaming . context import StreamingContext <EOL> from pyspark_cassandra import CassandraSparkContext , RowFormat , Row , UDT <EOL> import pyspark_cassandra <EOL> import pyspark_cassandra . streaming <EOL> from pyspark_cassandra . conf import ReadConf , WriteConf <EOL> from itertools import chain <EOL> from math import sqrt <EOL> from uuid import UUID <EOL> class CassandraTestCase ( unittest . TestCase ) : <EOL> keyspace = "<STR_LIT>" <EOL> def rdd ( self , keyspace = None , table = None , key = None , column = None , ** kwargs ) : <EOL> keyspace = keyspace or getattr ( self , '<STR_LIT>' , None ) <EOL> table = table or getattr ( self , '<STR_LIT>' , None ) <EOL> rdd = self . sc . cassandraTable ( keyspace , table , ** kwargs ) <EOL> if key is not None : <EOL> rdd = rdd . where ( '<STR_LIT>' , key ) <EOL> if column is not None : <EOL> rdd = rdd . select ( column ) <EOL> return rdd <EOL> def read_test ( self , type_name , value = None ) : <EOL> rdd = self . rdd ( key = type_name , column = type_name ) <EOL> self . assertEqual ( rdd . count ( ) , <NUM_LIT:1> ) <EOL> read = getattr ( rdd . first ( ) , type_name ) <EOL> self . assertEqual ( read , value ) <EOL> return read <EOL> def read_write_test ( self , type_name , value ) : <EOL> row = { '<STR_LIT:key>' : type_name , type_name : value } <EOL> rdd = self . sc . parallelize ( [ row ] ) <EOL> rdd . saveToCassandra ( self . keyspace , self . table ) <EOL> return self . read_test ( type_name , value ) <EOL> class SimpleTypesTestBase ( CassandraTestCase ) : <EOL> table = "<STR_LIT>" <EOL> simple_types = [ <EOL> '<STR_LIT:ascii>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:float>' , <EOL> '<STR_LIT>' , '<STR_LIT:int>' , '<STR_LIT:text>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( SimpleTypesTestBase , cls ) . setUpClass ( ) <EOL> cls . session . execute ( '''<STR_LIT>''' + cls . table + '''<STR_LIT>''' % '<STR_LIT:U+002CU+0020>' . join ( '<STR_LIT>' . format ( t ) for t in cls . simple_types ) ) <EOL> def setUp ( self ) : <EOL> super ( SimpleTypesTestBase , self ) . setUp ( ) <EOL> self . session . execute ( '<STR_LIT>' + self . table ) <EOL> class SimpleTypesTest ( SimpleTypesTestBase ) : <EOL> def test_ascii ( self ) : <EOL> self . read_write_test ( '<STR_LIT:ascii>' , '<STR_LIT>' ) <EOL> def test_bigint ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , sys . maxint ) <EOL> def test_blob ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , bytearray ( '<STR_LIT>' ) ) <EOL> def test_boolean ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , False ) <EOL> def test_decimal ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , Decimal ( <NUM_LIT:0.5> ) ) <EOL> def test_double ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , <NUM_LIT:0.5> ) <EOL> def test_float ( self ) : <EOL> self . read_write_test ( '<STR_LIT:float>' , <NUM_LIT:0.5> ) <EOL> def test_int ( self ) : <EOL> self . read_write_test ( '<STR_LIT:int>' , <NUM_LIT:1> ) <EOL> def test_text ( self ) : <EOL> self . read_write_test ( '<STR_LIT:text>' , u'<STR_LIT>' ) <EOL> def test_timeuuid ( self ) : <EOL> uuid = uuid_from_time ( datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> ) ) <EOL> self . read_write_test ( '<STR_LIT>' , uuid ) <EOL> def test_varchar ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , u'<STR_LIT>' ) <EOL> def test_varint ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> def test_uuid ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , uuid . UUID ( '<STR_LIT>' ) ) <EOL> class CollectionTypesTest ( CassandraTestCase ) : <EOL> table = "<STR_LIT>" <EOL> collection_types = { <EOL> '<STR_LIT:m>' : '<STR_LIT>' , <EOL> '<STR_LIT:l>' : '<STR_LIT>' , <EOL> '<STR_LIT:s>' : '<STR_LIT>' , <EOL> } <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( CollectionTypesTest , cls ) . setUpClass ( ) <EOL> cls . session . execute ( '''<STR_LIT>''' % ( cls . table , '<STR_LIT:U+002CU+0020>' . join ( '<STR_LIT>' % ( k , v ) for k , v in cls . collection_types . items ( ) ) ) ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> super ( CollectionTypesTest , cls ) . tearDownClass ( ) <EOL> def setUp ( self ) : <EOL> super ( CollectionTypesTest , self ) . setUp ( ) <EOL> self . session . execute ( '<STR_LIT>' % self . table ) <EOL> def collections_common_tests ( self , collection , column ) : <EOL> rows = [ <EOL> { '<STR_LIT:key>' : k , column : v } <EOL> for k , v in collection . items ( ) <EOL> ] <EOL> self . sc . parallelize ( rows ) . saveToCassandra ( self . keyspace , self . table ) <EOL> rdd = self . sc . cassandraTable ( self . keyspace , self . table ) . select ( '<STR_LIT:key>' , column ) . cache ( ) <EOL> self . assertEqual ( len ( collection ) , rdd . count ( ) ) <EOL> collected = rdd . collect ( ) <EOL> self . assertEqual ( len ( collection ) , len ( collected ) ) <EOL> for row in collected : <EOL> self . assertEqual ( collection [ row . key ] , getattr ( row , column ) ) <EOL> return rdd <EOL> def test_list ( self ) : <EOL> lists = { '<STR_LIT>' % i : list ( string . ascii_lowercase [ : i ] ) for i in range ( <NUM_LIT:1> , <NUM_LIT:10> ) } <EOL> self . collections_common_tests ( lists , '<STR_LIT:l>' ) <EOL> def test_map ( self ) : <EOL> maps = { '<STR_LIT>' % i : { k : '<STR_LIT:x>' for k in string . ascii_lowercase [ : i ] } for i in range ( <NUM_LIT:1> , <NUM_LIT:10> ) } <EOL> self . collections_common_tests ( maps , '<STR_LIT:m>' ) <EOL> def test_set ( self ) : <EOL> maps = { '<STR_LIT>' % i : set ( string . ascii_lowercase [ : i ] ) for i in range ( <NUM_LIT:1> , <NUM_LIT:10> ) } <EOL> self . collections_common_tests ( maps , '<STR_LIT:s>' ) <EOL> class UDTTest ( CassandraTestCase ) : <EOL> table = "<STR_LIT>" <EOL> types = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:text>' , <EOL> '<STR_LIT>' : '<STR_LIT:int>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:text>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( UDTTest , cls ) . setUpClass ( ) <EOL> cls . udt_support = cls . session . cluster . protocol_version >= <NUM_LIT:4> <EOL> if cls . udt_support : <EOL> for name , udt in cls . types . items ( ) : <EOL> cls . session . execute ( '''<STR_LIT>''' % ( name , '<STR_LIT>' . join ( '<STR_LIT>' % f for f in udt . items ( ) ) ) ) <EOL> fields = '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' . format ( udt_type = udt_type ) <EOL> for udt_type in cls . types <EOL> ) <EOL> fields += '<STR_LIT:U+002CU+0020>' + '<STR_LIT:U+002CU+0020>' . join ( <EOL> '<STR_LIT>' . format ( udt_type = udt_type , col_type = col_type ) <EOL> for udt_type in cls . types <EOL> for col_type in ( '<STR_LIT>' , '<STR_LIT:list>' ) <EOL> ) <EOL> cls . session . execute ( '''<STR_LIT>''' % ( cls . table , fields ) ) <EOL> def setUp ( self ) : <EOL> if not self . udt_support : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> super ( UDTTest , self ) . setUp ( ) <EOL> self . session . execute ( '<STR_LIT>' % self . table ) <EOL> def read_write_test ( self , type_name , value ) : <EOL> read = super ( UDTTest , self ) . read_write_test ( type_name , value ) <EOL> self . assertTrue ( isinstance ( read , UDT ) , <EOL> '<STR_LIT>' ) <EOL> udt = self . types [ type_name ] <EOL> for field in udt : <EOL> self . assertEqual ( getattr ( read , field ) , value [ field ] ) <EOL> def test_simple_udt ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , UDT ( col_text = '<STR_LIT:text>' , col_int = <NUM_LIT:1> , col_boolean = True ) ) <EOL> def test_simple_udt_null ( self ) : <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , None ) <EOL> def test_simple_udt_null_field ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , UDT ( col_text = '<STR_LIT:text>' , col_int = None , col_boolean = True ) ) <EOL> self . read_write_test ( '<STR_LIT>' , UDT ( col_text = None , col_int = <NUM_LIT:1> , col_boolean = True ) ) <EOL> def test_udt_wset ( self ) : <EOL> self . read_write_test ( '<STR_LIT>' , UDT ( col_text = '<STR_LIT:text>' , col_set = { <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> } ) ) <EOL> def test_collection_of_udts ( self ) : <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , None ) <EOL> udts = [ UDT ( col_text = '<STR_LIT>' + str ( i ) , col_int = i , col_boolean = bool ( i % <NUM_LIT:2> ) ) for i in range ( <NUM_LIT:10> ) ] <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , set ( udts ) ) <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , udts ) <EOL> udts = [ UDT ( col_text = '<STR_LIT>' + str ( i ) , col_int = i , col_boolean = None ) for i in range ( <NUM_LIT:10> ) ] <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , set ( udts ) ) <EOL> super ( UDTTest , self ) . read_write_test ( '<STR_LIT>' , udts ) <EOL> class SelectiveSaveTest ( SimpleTypesTestBase ) : <EOL> def _save_and_get ( self , * row ) : <EOL> columns = [ '<STR_LIT:key>' , '<STR_LIT:text>' ] <EOL> self . sc . parallelize ( row ) . saveToCassandra ( self . keyspace , self . table , columns = columns ) <EOL> rdd = self . rdd ( ) . select ( * columns ) <EOL> self . assertEqual ( rdd . count ( ) , <NUM_LIT:1> ) <EOL> return rdd . first ( ) <EOL> def test_row ( self ) : <EOL> row = Row ( key = '<STR_LIT>' , int = <NUM_LIT:2> , text = '<STR_LIT:a>' , boolean = False ) <EOL> read = self . _save_and_get ( row ) <EOL> for k in [ '<STR_LIT:key>' , '<STR_LIT:text>' ] : <EOL> self . assertEqual ( getattr ( row , k ) , getattr ( read , k ) ) <EOL> for k in [ '<STR_LIT>' , '<STR_LIT:int>' ] : <EOL> self . assertIsNone ( getattr ( read , k , None ) ) <EOL> def test_dict ( self ) : <EOL> row = dict ( key = '<STR_LIT>' , int = <NUM_LIT:2> , text = '<STR_LIT:a>' , boolean = False ) <EOL> read = self . _save_and_get ( row ) <EOL> for k in [ '<STR_LIT:key>' , '<STR_LIT:text>' ] : <EOL> self . assertEqual ( row [ k ] , read [ k ] ) <EOL> for k in [ '<STR_LIT>' , '<STR_LIT:int>' ] : <EOL> self . assertIsNone ( getattr ( read , k , None ) ) <EOL> class LimitAndTakeTest ( SimpleTypesTestBase ) : <EOL> size = <NUM_LIT:1000> <EOL> def setUp ( self ) : <EOL> super ( LimitAndTakeTest , self ) . setUp ( ) <EOL> data = self . sc . parallelize ( range ( <NUM_LIT:0> , self . size ) ) . map ( lambda i : { '<STR_LIT:key>' : i , '<STR_LIT:int>' : i } ) <EOL> data . saveToCassandra ( self . keyspace , self . table ) <EOL> def test_limit ( self ) : <EOL> data = self . rdd ( ) <EOL> for i in ( <NUM_LIT:5> , <NUM_LIT:10> , <NUM_LIT:100> , <NUM_LIT:1000> , <NUM_LIT> ) : <EOL> l = min ( i , self . size ) <EOL> self . assertEqual ( len ( data . take ( i ) ) , l ) <EOL> self . assertEqual ( len ( data . limit ( i ) . collect ( ) ) , l ) <EOL> self . assertEqual ( len ( data . limit ( i * <NUM_LIT:2> ) . take ( i ) ) , l ) <EOL> class FormatTest ( SimpleTypesTestBase ) : <EOL> expected = Row ( key = '<STR_LIT>' , int = <NUM_LIT:2> , text = '<STR_LIT:a>' ) <EOL> def setUp ( self ) : <EOL> super ( FormatTest , self ) . setUp ( ) <EOL> self . sc . parallelize ( [ self . expected ] ) . saveToCassandra ( self . keyspace , self . table ) <EOL> def read_as ( self , row_format , keyed ) : <EOL> table = self . rdd ( row_format = row_format ) <EOL> if keyed : <EOL> table = table . by_primary_key ( ) <EOL> table = table . where ( '<STR_LIT>' , self . expected . key ) <EOL> return table . first ( ) <EOL> def assert_rowtype ( self , row_format , row_type , keyed = False ) : <EOL> row = self . read_as ( row_format , keyed ) <EOL> self . assertEqual ( type ( row ) , row_type ) <EOL> return row <EOL> def assert_kvtype ( self , row_format , kv_type ) : <EOL> row = self . assert_rowtype ( row_format , tuple , keyed = True ) <EOL> self . assertEqual ( len ( row ) , <NUM_LIT:2> ) <EOL> k , v = row <EOL> self . assertEqual ( type ( k ) , kv_type ) <EOL> self . assertEqual ( type ( v ) , kv_type ) <EOL> return k , v <EOL> def test_tuple ( self ) : <EOL> row = self . assert_rowtype ( RowFormat . TUPLE , tuple ) <EOL> self . assertEqual ( self . expected . key , row [ <NUM_LIT:0> ] ) <EOL> def test_kvtuple ( self ) : <EOL> k , _ = self . assert_kvtype ( RowFormat . TUPLE , tuple ) <EOL> self . assertEqual ( self . expected . key , k [ <NUM_LIT:0> ] ) <EOL> def test_dict ( self ) : <EOL> row = self . assert_rowtype ( RowFormat . DICT , dict ) <EOL> self . assertEqual ( self . expected . key , row [ '<STR_LIT:key>' ] ) <EOL> def test_kvdict ( self ) : <EOL> k , _ = self . assert_kvtype ( RowFormat . DICT , dict ) <EOL> self . assertEqual ( self . expected . key , k [ '<STR_LIT:key>' ] ) <EOL> def test_row ( self ) : <EOL> row = self . assert_rowtype ( RowFormat . ROW , pyspark_cassandra . Row ) <EOL> self . assertEqual ( self . expected . key , row . key ) <EOL> def test_kvrow ( self ) : <EOL> k , _ = self . assert_kvtype ( RowFormat . ROW , pyspark_cassandra . Row ) <EOL> self . assertEqual ( self . expected . key , k . key ) <EOL> class ConfTest ( SimpleTypesTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( SimpleTypesTestBase , self ) . setUp ( ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> self . session . execute ( <EOL> "<STR_LIT>" <EOL> % ( self . table , i , i , i ) <EOL> ) <EOL> def test_read_conf ( self ) : <EOL> self . rdd ( split_count = <NUM_LIT:100> ) . collect ( ) <EOL> self . rdd ( split_size = <NUM_LIT:32> ) . collect ( ) <EOL> self . rdd ( fetch_size = <NUM_LIT:100> ) . collect ( ) <EOL> self . rdd ( consistency_level = '<STR_LIT>' ) . collect ( ) <EOL> self . rdd ( consistency_level = ConsistencyLevel . LOCAL_QUORUM ) . collect ( ) <EOL> self . rdd ( metrics_enabled = True ) . collect ( ) <EOL> self . rdd ( read_conf = ReadConf ( split_count = <NUM_LIT:10> , consistency_level = '<STR_LIT>' ) ) . collect ( ) <EOL> self . rdd ( read_conf = ReadConf ( consistency_level = '<STR_LIT>' , metrics_enabled = True ) ) . collect ( ) <EOL> def test_write_conf ( self ) : <EOL> rdd = self . sc . parallelize ( [ { '<STR_LIT:key>' : i , '<STR_LIT:text>' : i , '<STR_LIT:int>' : i } for i in range ( <NUM_LIT:10> ) ] ) <EOL> save = partial ( rdd . saveToCassandra , self . keyspace , self . table ) <EOL> save ( batch_size = <NUM_LIT:100> ) <EOL> save ( batch_buffer_size = <NUM_LIT:100> ) <EOL> save ( batch_grouping_key = '<STR_LIT>' ) <EOL> save ( batch_grouping_key = '<STR_LIT>' ) <EOL> save ( consistency_level = '<STR_LIT>' ) <EOL> save ( consistency_level = ConsistencyLevel . LOCAL_QUORUM ) <EOL> save ( parallelism_level = <NUM_LIT:10> ) <EOL> save ( throughput_mibps = <NUM_LIT:10> ) <EOL> save ( ttl = <NUM_LIT:5> ) <EOL> save ( ttl = timedelta ( minutes = <NUM_LIT:30> ) ) <EOL> save ( timestamp = time . clock ( ) * <NUM_LIT:1000> * <NUM_LIT:1000> ) <EOL> save ( timestamp = datetime . now ( ) ) <EOL> save ( metrics_enabled = True ) <EOL> save ( write_conf = WriteConf ( ttl = <NUM_LIT:3> , metrics_enabled = True ) ) <EOL> class StreamingTest ( SimpleTypesTestBase ) : <EOL> interval = <NUM_LIT> <EOL> size = <NUM_LIT:10> <EOL> count = <NUM_LIT:3> <EOL> rows = [ <EOL> [ <EOL> { '<STR_LIT:key>' : str ( j * size + i ) , '<STR_LIT:text>' : str ( j * size + i ) } <EOL> for i in range ( size ) <EOL> ] <EOL> for j in range ( count ) <EOL> ] <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( StreamingTest , cls ) . setUpClass ( ) <EOL> cls . ssc = StreamingContext ( cls . sc , cls . interval ) <EOL> def setUp ( self ) : <EOL> super ( StreamingTest , self ) . setUp ( ) <EOL> self . rdds = list ( map ( self . sc . parallelize , self . rows ) ) <EOL> self . stream = self . ssc . queueStream ( self . rdds ) <EOL> def test ( self ) : <EOL> self . stream . saveToCassandra ( self . keyspace , self . table ) <EOL> self . ssc . start ( ) <EOL> self . ssc . awaitTermination ( ( self . count + <NUM_LIT:1> ) * self . interval ) <EOL> self . ssc . stop ( stopSparkContext = False , stopGraceFully = True ) <EOL> tbl = self . rdd ( row_format = RowFormat . TUPLE ) . select ( '<STR_LIT:key>' , '<STR_LIT:text>' ) <EOL> read = tbl . by_primary_key ( ) . collect ( ) <EOL> self . assertEqual ( len ( read ) , self . size * self . count ) <EOL> for ( k , v ) in read : <EOL> self . assertEqual ( k , v ) <EOL> class JoinRDDTest ( SimpleTypesTestBase ) : <EOL> def setUp ( self ) : <EOL> super ( JoinRDDTest , self ) . setUp ( ) <EOL> def test_simple_pk ( self ) : <EOL> table = '<STR_LIT>' <EOL> self . session . execute ( '''<STR_LIT>''' + table + '''<STR_LIT>''' ) <EOL> self . session . execute ( '<STR_LIT>' % table ) <EOL> rows = { <EOL> str ( c ) : str ( i ) for i , c in <EOL> enumerate ( string . ascii_lowercase ) <EOL> } <EOL> for k , v in rows . items ( ) : <EOL> self . session . execute ( <EOL> '<STR_LIT>' + table + '<STR_LIT>' , ( k , v ) <EOL> ) <EOL> rdd = self . sc . parallelize ( rows . items ( ) ) <EOL> self . assertEqual ( dict ( rdd . collect ( ) ) , rows ) <EOL> tbl = rdd . joinWithCassandraTable ( self . keyspace , table ) <EOL> joined = tbl . on ( '<STR_LIT:key>' ) . select ( '<STR_LIT:key>' , '<STR_LIT:value>' ) . cache ( ) <EOL> self . assertEqual ( dict ( joined . keys ( ) . collect ( ) ) , dict ( joined . values ( ) . collect ( ) ) ) <EOL> for ( k , v ) in joined . collect ( ) : <EOL> self . assertEqual ( k , v ) <EOL> def test_composite_pk ( self ) : <EOL> table = '<STR_LIT>' <EOL> self . session . execute ( '''<STR_LIT>''' + table + '''<STR_LIT>''' ) <EOL> self . session . execute ( '<STR_LIT>' % table ) <EOL> rows = [ <EOL> ( unicode ( pk ) , unicode ( cc ) , unicode ( pk + '<STR_LIT:->' + cc ) ) <EOL> for pk in string . ascii_lowercase [ : <NUM_LIT:3> ] <EOL> for cc in ( str ( i ) for i in range ( <NUM_LIT:3> ) ) <EOL> ] <EOL> for row in rows : <EOL> self . session . execute ( <EOL> '<STR_LIT>' + table + '<STR_LIT>' , row <EOL> ) <EOL> rdd = self . sc . parallelize ( rows ) <EOL> joined = rdd . joinWithCassandraTable ( self . keyspace , table ) . on ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( sorted ( zip ( rows , rows ) ) , sorted ( joined . map ( tuple ) . collect ( ) ) ) <EOL> joined = rdd . joinWithCassandraTable ( self . keyspace , table ) . on ( '<STR_LIT>' ) <EOL> self . assertEqual ( len ( rows ) * sqrt ( len ( rows ) ) , joined . count ( ) ) <EOL> class JoinDStreamTest ( StreamingTest ) : <EOL> def setUp ( self ) : <EOL> super ( JoinDStreamTest , self ) . setUp ( ) <EOL> self . joined_rows = self . sc . accumulator ( [ ] , accum_param = AddingAccumulatorParam ( [ ] ) ) <EOL> def checkRDD ( self , time , rdd ) : <EOL> self . joined_rows += rdd . collect ( ) <EOL> def test ( self ) : <EOL> rows = list ( chain ( * self . rows ) ) <EOL> rows_by_key = { row [ '<STR_LIT:key>' ] : row for row in rows } <EOL> self . sc . parallelize ( rows ) . saveToCassandra ( self . keyspace , self . table ) <EOL> self . stream . joinWithCassandraTable ( self . keyspace , self . table , [ '<STR_LIT:text>' ] , [ '<STR_LIT:key>' ] ) . foreachRDD ( self . checkRDD ) <EOL> self . ssc . start ( ) <EOL> self . ssc . awaitTermination ( ( self . count + <NUM_LIT:1> ) * self . interval ) <EOL> self . ssc . stop ( stopSparkContext = False , stopGraceFully = True ) <EOL> joined_rows = self . joined_rows . value <EOL> self . assertEqual ( len ( joined_rows ) , len ( rows ) ) <EOL> for row in joined_rows : <EOL> self . assertEqual ( len ( row ) , <NUM_LIT:2> ) <EOL> left , right = row <EOL> self . assertEqual ( type ( left ) , type ( right ) ) <EOL> self . assertEqual ( rows_by_key [ left [ '<STR_LIT:key>' ] ] , left ) <EOL> self . assertEqual ( left [ '<STR_LIT:text>' ] , right [ '<STR_LIT:text>' ] ) <EOL> self . assertEqual ( len ( right ) , <NUM_LIT:1> ) <EOL> class RegressionTest ( CassandraTestCase ) : <EOL> def test_64 ( self ) : <EOL> self . session . execute ( '''<STR_LIT>''' ) <EOL> self . session . execute ( '''<STR_LIT>''' ) <EOL> res = ( [ <NUM_LIT:0.0> , <NUM_LIT:1.0> , <NUM_LIT> ] , [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT:0.0> ] , <NUM_LIT:0.0> ) <EOL> rdd = self . sc . parallelize ( [ res ] ) <EOL> rdd . saveToCassandra ( self . keyspace , '<STR_LIT>' , columns = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> row = self . rdd ( table = '<STR_LIT>' ) . first ( ) <EOL> self . assertEqual ( row . pos , res [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( row . pdf , res [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( row . delay , res [ <NUM_LIT:2> ] ) <EOL> def test_89 ( self ) : <EOL> self . session . execute ( '''<STR_LIT>''' ) <EOL> self . session . execute ( '''<STR_LIT>''' ) <EOL> self . sc . parallelize ( [ dict ( id = '<STR_LIT:a>' , val = '<STR_LIT:b>' ) ] ) . saveToCassandra ( self . keyspace , '<STR_LIT>' ) <EOL> joined = ( self . sc <EOL> . parallelize ( [ dict ( id = '<STR_LIT:a>' , uuid = UUID ( '<STR_LIT>' ) ) ] ) <EOL> . joinWithCassandraTable ( self . keyspace , '<STR_LIT>' ) <EOL> . collect ( ) <EOL> ) <EOL> self . assertEqual ( len ( joined ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( joined [ <NUM_LIT:0> ] ) , <NUM_LIT:2> ) <EOL> left , right = joined [ <NUM_LIT:0> ] <EOL> self . assertEqual ( left [ '<STR_LIT:id>' ] , '<STR_LIT:a>' ) <EOL> self . assertEqual ( left [ '<STR_LIT>' ] , UUID ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( right [ '<STR_LIT:id>' ] , '<STR_LIT:a>' ) <EOL> self . assertEqual ( right [ '<STR_LIT>' ] , '<STR_LIT:b>' ) <EOL> def test_93 ( self ) : <EOL> self . session . execute ( '''<STR_LIT>''' ) <EOL> self . sc . parallelize ( [ <EOL> Row ( name = str ( i ) , data_final = bytearray ( str ( i ) ) , data_inter = bytearray ( str ( i ) ) , <EOL> family_label = str ( i ) , rand = i / <NUM_LIT:10> , source = str ( i ) , score = i * <NUM_LIT:10> ) <EOL> for i in range ( <NUM_LIT:4> ) <EOL> ] ) . saveToCassandra ( self . keyspace , '<STR_LIT>' ) <EOL> joined = ( self . sc <EOL> . parallelize ( [ <EOL> Row ( name = '<STR_LIT:1>' , score = <NUM_LIT> ) , <EOL> Row ( name = '<STR_LIT:2>' , score = <NUM_LIT:0.5> ) , <EOL> ] ) <EOL> . joinWithCassandraTable ( self . keyspace , '<STR_LIT>' ) <EOL> . on ( '<STR_LIT:name>' ) . collect ( ) <EOL> ) <EOL> self . assertEqual ( len ( joined ) , <NUM_LIT:2> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> try : <EOL> CassandraTestCase . session = Cluster ( ) . connect ( ) <EOL> CassandraTestCase . session . execute ( '''<STR_LIT>''' % ( CassandraTestCase . keyspace , ) ) <EOL> CassandraTestCase . session . set_keyspace ( CassandraTestCase . keyspace ) <EOL> CassandraTestCase . sc = CassandraSparkContext ( conf = SparkConf ( ) . setAppName ( "<STR_LIT>" ) ) <EOL> unittest . main ( ) <EOL> finally : <EOL> CassandraTestCase . sc . stop ( ) <EOL> CassandraTestCase . session . shutdown ( ) </s>
