<s> from __future__ import print_function , absolute_import , division <EOL> import pytest <EOL> pytest . importorskip ( '<STR_LIT>' ) <EOL> import pytest <EOL> from datashape import dshape <EOL> from odo import odo , discover <EOL> from odo . backends . sparksql import SparkDataFrame <EOL> from pyspark import RDD <EOL> from pyspark . rdd import PipelinedRDD <EOL> from pyspark . sql import Row , SchemaRDD <EOL> data = [ [ '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:1> ] , <EOL> [ '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:2> ] , <EOL> [ '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:3> ] ] <EOL> @ pytest . fixture <EOL> def rdd ( sc ) : <EOL> return sc . parallelize ( data ) <EOL> def test_spark_into ( rdd ) : <EOL> with pytest . raises ( NotImplementedError ) : <EOL> odo ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , rdd ) <EOL> def test_spark_into_context ( sc ) : <EOL> seq = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> rdd = odo ( seq , sc ) <EOL> assert isinstance ( rdd , RDD ) <EOL> assert odo ( rdd , [ ] ) == seq <EOL> def test_rdd_into_schema_rdd ( rdd ) : <EOL> ds = dshape ( '<STR_LIT>' ) <EOL> srdd = odo ( rdd , SparkDataFrame , dshape = ds ) <EOL> assert isinstance ( srdd , ( SparkDataFrame , SchemaRDD ) ) <EOL> assert list ( map ( set , srdd . collect ( ) ) ) == list ( map ( set , rdd . collect ( ) ) ) <EOL> def test_pipelined_rdd_into_schema_rdd ( rdd ) : <EOL> pipelined = rdd . map ( lambda x : Row ( amount = x [ <NUM_LIT:1> ] ) ) <EOL> assert isinstance ( pipelined , PipelinedRDD ) <EOL> srdd = odo ( pipelined , SparkDataFrame , <EOL> dshape = dshape ( '<STR_LIT>' ) ) <EOL> assert isinstance ( srdd , ( SparkDataFrame , SchemaRDD ) ) <EOL> assert ( list ( map ( set , srdd . collect ( ) ) ) == <EOL> list ( map ( set , pipelined . collect ( ) ) ) ) <EOL> def test_discover_rdd ( rdd ) : <EOL> assert discover ( rdd ) . subshape [ <NUM_LIT:0> ] == discover ( data ) . subshape [ <NUM_LIT:0> ] </s>
