Welcome to sparkly’s documentation!¶

Installing dependencies¶

Why: Spark forces you to specify dependencies (spark packages or maven artifacts) when a spark job is submitted (something like spark-submit --packages=...). We prefer a code-first approach where dependencies are actually declared as part of the job.

For example: You want to read data from Cassandra.

from sparkly import SparklySession


class MySession(SparklySession):
    # Define a list of spark packages or maven artifacts.
    packages = [
        'datastax:spark-cassandra-connector:2.0.0-M2-s_2.11',
    ]

# Dependencies will be fetched during the session initialisation.
spark = MySession()

# Here is how you now can access a dataset in Cassandra.
df = spark.read_ext.by_url('cassandra://<cassandra-host>/<db>/<table>?consistency=QUORUM')

Custom Maven repositories¶

Why: If you have a private maven repository, this is how to point spark to it when it performs a package lookup. Order in which dependencies will be resolved is next:

• Local cache
• Custom maven repositories (if specified)
• Maven Central

For example: Let’s assume your maven repository is available on: http://my.repo.net/maven, and there is some spark package published there, with identifier: my.corp:spark-handy-util:0.0.1 You can install it to a spark session like this:

..code-block:: python

from sparkly import SparklySession

class MySession(SparklySession):

repositories = [‘http://my.repo.net/maven‘] packages = [‘my.corp:spark-handy-util:0.0.1’]

spark = MySession()

Tuning options¶

Why: You want to customise your spark session.

For example:

• spark.sql.shuffle.partitions to tune shuffling;
• hive.metastore.uris to connect to your own HiveMetastore;
• spark.hadoop.avro.mapred.ignore.inputs.without.extension package specific options.

from sparkly import SparklySession


class MySession(SparklySession):
    options = {
        # Increase the default amount of partitions for shuffling.
        'spark.sql.shuffle.partitions': 1000,
        # Setup remote Hive Metastore.
        'hive.metastore.uris': 'thrift://<host1>:9083,thrift://<host2>:9083',
        # Ignore files without `avro` extensions.
        'spark.hadoop.avro.mapred.ignore.inputs.without.extension': 'false',
    }

# You can also overwrite or add some options at initialisation time.
spark = MySession({'spark.sql.shuffle.partitions': 10})

Using UDFs¶

Why: To start using Java UDF you have to import JAR file via SQL query like add jar ../path/to/file and then call registerJavaFunction. We think it’s too many actions for such simple functionality.

For example: You want to import UDFs from brickhouse library.

from pyspark.sql.types import IntegerType
from sparkly import SparklySession


def my_own_udf(item):
    return len(item)


class MySession(SparklySession):
    # Import local jar files.
    jars = [
        '/path/to/brickhouse.jar'
    ]
    # Define UDFs.
    udfs = {
        'collect_max': 'brickhouse.udf.collect.CollectMaxUDAF',  # Java UDF.
        'my_udf': (my_own_udf, IntegerType()),  # Python UDF.
    }

spark = MySession()

spark.sql('SELECT collect_max(amount) FROM my_data GROUP BY ...')
spark.sql('SELECT my_udf(amount) FROM my_data')

API documentation¶

class sparkly.session.SparklySession(additional_options=None)[source]¶

Wrapper around HiveContext to simplify definition of options, packages, JARs and UDFs.

Example:

from pyspark.sql.types import IntegerType
import sparkly


class MySession(sparkly.SparklySession):
    options = {'spark.sql.shuffle.partitions': '2000'}
    repositories = ['http://packages.confluent.io/maven/']
    packages = ['com.databricks:spark-csv_2.10:1.4.0']
    jars = ['../path/to/brickhouse-0.7.1.jar']
    udfs = {
        'collect_max': 'brickhouse.udf.collect.CollectMaxUDAF',
        'my_python_udf': (lambda x: len(x), IntegerType()),
    }


spark = MySession()
spark.read_ext.cassandra(...)

options¶

dict[str,str] – Configuration options that are passed to SparkConf. See the list of possible options.

repositories¶

list[str] – List of additional maven repositories for package lookup.

packages¶

list[str] – Spark packages that should be installed. See https://spark-packages.org/

jars¶

list[str] – Full paths to jar files that we want to include to the session. E.g. a JDBC connector or a library with UDF functions.

udfs¶

dict[str,str|typing.Callable] – Register UDF functions within the session. Key - a name of the function, Value - either a class name imported from a JAR file

or a tuple with python function and its return type.

has_jar(jar_name)[source]¶

Check if the jar is available in the session.

Parameters:	jar_name (str) – E.g. “mysql-connector-java”.
Returns:	bool

has_package(package_prefix)[source]¶

Check if the package is available in the session.

Parameters:	package_prefix (str) – E.g. “org.elasticsearch:elasticsearch-spark”.
Returns:	bool

Cassandra¶

Sparkly relies on the official spark cassandra connector and was successfully tested in production using version 2.0.0-M2.

from sparkly import SparklySession


class MySession(SparklySession):
    # Feel free to play with other versions
    packages = ['datastax:spark-cassandra-connector:2.0.0-M2-s_2.11']

spark = MySession()

# To read data
df = spark.read_ext.cassandra('localhost', 'my_keyspace', 'my_table')
# To write data
df.write_ext.cassandra('localhost', 'my_keyspace', 'my_table')

Elastic¶

Sparkly relies on the official elastic spark connector and was successfully tested in production using version 5.1.1.

from sparkly import SparklySession


class MySession(SparklySession):
    # Feel free to play with other versions
    packages = ['org.elasticsearch:elasticsearch-spark-20_2.11:5.1.1']

spark = MySession()

# To read data
df = spark.read_ext.elastic('localhost', 'my_index', 'my_type', query='?q=awesomeness')
# To write data
df.write_ext.elastic('localhost', 'my_index', 'my_type')

Kafka¶

Sparkly’s reader and writer for Kafka are built on top of the official spark package for Kafka and python library kafka-python . The first one allows us to read data efficiently, the second covers a lack of writing functionality in the official distribution.

import json

from sparkly import SparklySession


class MySession(SparklySession):
    packages = [
        'org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.0',
    ]

spark = MySession()

# To read JSON messaged from Kafka into a dataframe:

#   1. Define a schema of the messages you read.
df_schema = StructType([
    StructField('key', StructType([
        StructField('id', StringType(), True)
    ])),
    StructField('value', StructType([
        StructField('name', StringType(), True),
        StructField('surname', StringType(), True),
    ]))
])

#   2. Specify the schema as a reader parameter.
df = hc.read_ext.kafka(
    'kafka.host',
    topic='my.topic',
    key_deserializer=lambda item: json.loads(item.decode('utf-8')),
    value_deserializer=lambda item: json.loads(item.decode('utf-8')),
    schema=df_schema,
)

# To write a dataframe to Kafka in JSON format:
df.write_ext.kafka(
    'kafka.host',
    topic='my.topic',
    key_serializer=lambda item: json.dumps(item).encode('utf-8'),
    value_serializer=lambda item: json.dumps(item).encode('utf-8'),
)

MySQL¶

Basically, it’s just a high level api on top of the native jdbc reader and jdbc writer.

from sparkly import SparklySession
from sparkly.utils import absolute_path


class MySession(SparklySession):
    # Feel free to play with other versions.
    packages = ['mysql:mysql-connector-java:5.1.39']


spark = MySession()

# To read data
df = spark.read_ext.mysql('localhost', 'my_database', 'my_table',
                          options={'user': 'root', 'password': 'root'})
# To write data
df.write_ext.mysql('localhost', 'my_database', 'my_table', options={
    'user': 'root',
    'password': 'root',
    'rewriteBatchedStatements': 'true',  # improves write throughput dramatically
})

Universal reader/writer¶

The DataFrame abstraction is really powerful when it comes to transformations. You can shape your data from various storages using exactly the same api. For instance, you can join data from Cassandra with data from Elasticsearch and write the result to MySQL.

The only problem - you have to explicitly define sources (or destinations) in order to create (or export) a DataFrame. But the source/destination of data doesn’t really change the logic of transformations (if the schema is preserved). To solve the problem, we decided to add the universal api to read/write DataFrames:

from sparkly import SparklyContext

class MyContext(SparklyContext):
    packages = [
        'datastax:spark-cassandra-connector:1.6.1-s_2.10',
        'com.databricks:spark-csv_2.10:1.4.0',
        'org.elasticsearch:elasticsearch-spark_2.10:2.3.0',
    ]

hc = MyContext()

# To read data
df = hc.read_ext.by_url('cassandra://localhost/my_keyspace/my_table?consistency=ONE')
df = hc.read_ext.by_url('csv:s3://my-bucket/my-data?header=true')
df = hc.read_ext.by_url('elastic://localhost/my_index/my_type?q=awesomeness')
df = hc.read_ext.by_url('parquet:hdfs://my.name.node/path/on/hdfs')

# To write data
df.write_ext.by_url('cassandra://localhost/my_keyspace/my_table?consistency=QUORUM&parallelism=8')
df.write_ext.by_url('csv:hdfs://my.name.node/path/on/hdfs')
df.write_ext.by_url('elastic://localhost/my_index/my_type?parallelism=4')
df.write_ext.by_url('parquet:s3://my-bucket/my-data?header=false')

Controlling the load¶

From the official documentation:

Don’t create too many partitions in parallel on a large cluster; otherwise Spark might crash your external database systems.

link: <https://spark.apache.org/docs/2.0.1/api/java/org/apache/spark/sql/DataFrameReader.html>

It’s a very good advice, but in practice it’s hard to track the number of partitions. For instance, if you write a result of a join operation to database the number of splits might be changed implicitly via spark.sql.shuffle.partitions.

To prevent us from shooting to the foot, we decided to add parallelism option for all our readers and writers. The option is designed to control a load on a source we write to / read from. It’s especially useful when you are working with data storages like Cassandra, MySQL or Elastic. However, the implementation of the throttling has some drawbacks and you should be aware of them.

The way we implemented it is pretty simple: we use coalesce on a dataframe to reduce an amount of tasks that will be executed in parallel. Let’s say you have a dataframe with 1000 splits and you want to write no more than 10 task in parallel. In such case coalesce will create a dataframe that has 10 splits with 100 original tasks in each. An outcome of this: if any of these 100 tasks fails, we have to retry the whole pack in 100 tasks.

Read more about coalesce

Reader API documentation¶

class sparkly.reader.SparklyReader(spark)[source]¶

A set of tools to create DataFrames from the external storages.

Note

This is a private class to the library. You should not use it directly. The instance of the class is available under SparklyContext via read_ext attribute.

by_url(url)[source]¶

Create a dataframe using url.

The main idea behind the method is to unify data access interface for different formats and locations. A generic schema looks like:

format:[protocol:]//host[:port][/location][?configuration]

Supported formats:

• CSV csv://
• Cassandra cassandra://
• Elastic elastic://
• MySQL mysql://
• Parquet parquet://
• Hive Metastore table table://

Query string arguments are passed as parameters to the relevant reader.

For instance, the next data source URL:

cassandra://localhost:9042/my_keyspace/my_table?consistency=ONE
    &parallelism=3&spark.cassandra.connection.compression=LZ4

Is an equivalent for:

hc.read_ext.cassandra(
    host='localhost',
    port=9042,
    keyspace='my_keyspace',
    table='my_table',
    consistency='ONE',
    parallelism=3,
    options={'spark.cassandra.connection.compression': 'LZ4'},
)

More examples:

table://table_name
csv:s3://some-bucket/some_directory?header=true
csv://path/on/local/file/system?header=false
parquet:s3://some-bucket/some_directory
elastic://elasticsearch.host/es_index/es_type?parallelism=8
cassandra://cassandra.host/keyspace/table?consistency=QUORUM
mysql://mysql.host/database/table

Parameters:	url (str) – Data source URL.
Returns:	pyspark.sql.DataFrame

cassandra(host, keyspace, table, consistency=None, port=None, parallelism=None, options=None)[source]¶

Create a dataframe from a Cassandra table.

Parameters:

host (str) – Cassandra server host. keyspace (str) – table (str) – Cassandra table to read from. consistency (str) – Read consistency level: ONE, QUORUM, ALL, etc. port (int|None) – Cassandra server port. parallelism (int|None) – The max number of parallel tasks that could be executed during the read stage (see Controlling the load). options (dict[str,str]|None) – Additional options for org.apache.spark.sql.cassandra format (see configuration for Cassandra).

Returns:

pyspark.sql.DataFrame

elastic(host, es_index, es_type, query='', fields=None, port=None, parallelism=None, options=None)[source]¶

Create a dataframe from an ElasticSearch index.

Parameters:

host (str) – Elastic server host. es_index (str) – Elastic index. es_type (str) – Elastic type. query (str) – Pre-filter es documents, e.g. ‘?q=views:>10’. fields (list[str]|None) – Select only specified fields. port (int|None) – parallelism (int|None) – The max number of parallel tasks that could be executed during the read stage (see Controlling the load). options (dict[str,str]) – Additional options for org.elasticsearch.spark.sql format (see configuration for Elastic).

Returns:

pyspark.sql.DataFrame

kafka(host, topic, offset_ranges=None, key_deserializer=None, value_deserializer=None, schema=None, port=9092, parallelism=None, options=None)[source]¶

Creates dataframe from specified set of messages from Kafka topic.

Defining ranges:

• If offset_ranges is specified it defines which specific range to read.
• If offset_ranges is omitted it will auto-discover it’s partitions.

The schema parameter, if specified, should contain two top level fields: key and value.

Parameters key_deserializer and value_deserializer are callables which get bytes as input and should return python structures as output.

Parameters:	host (str) – Kafka host. topic (str|None) – Kafka topic to read from. offset_ranges (list[(int, int, int) – List of partition ranges [(partition, start_offset, end_offset)]. key_deserializer (function) – Function used to deserialize the key. value_deserializer (function) – Function used to deserialize the value. schema (pyspark.sql.types.StructType) – Schema to apply to create a Dataframe. port (int) – Kafka port. parallelism (int|None) – The max number of parallel tasks that could be executed during the read stage (see Controlling the load). options (dict|None) – Additional kafka parameters, see KafkaUtils.createRDD docs.
Returns:	pyspark.sql.DataFrame
Raises:	InvalidArgumentError

mysql(host, database, table, port=None, parallelism=None, options=None)[source]¶

Create a dataframe from a MySQL table.

Options should include user and password.

Parameters:	host (str) – MySQL server address. database (str) – Database to connect to. table (str) – Table to read rows from. port (int|None) – MySQL server port. parallelism (int|None) – The max number of parallel tasks that could be executed during the read stage (see Controlling the load). options (dict[str,str]|None) – Additional options for JDBC reader (see configuration for MySQL).
Returns:	pyspark.sql.DataFrame

Writer API documentation¶

class sparkly.writer.SparklyWriter(df)[source]¶

A set of tools to write DataFrames to external storages.

Note

We don’t expect you to be using the class directly. The instance of the class is available under DataFrame via write_ext attribute.

by_url(url)[source]¶

Write a dataframe to a destination specified by url.

The main idea behind the method is to unify data export interface for different formats and locations. A generic schema looks like:

format:[protocol:]//host[:port][/location][?configuration]

Supported formats:

• CSV csv://
• Cassandra cassandra://
• Elastic elastic://
• MySQL mysql://
• Parquet parquet://

Query string arguments are passed as parameters to the relevant writer.

For instance, the next data export URL:

elastic://localhost:9200/my_index/my_type?&parallelism=3&mode=overwrite
    &es.write.operation=upsert

Is an equivalent for:

hc.read_ext.elastic(
    host='localhost',
    port=9200,
    es_index='my_index',
    es_type='my_type',
    parallelism=3,
    mode='overwrite',
    options={'es.write.operation': 'upsert'},
)

More examples:

csv:s3://some-s3-bucket/some-s3-key?partitionBy=date,platform
cassandra://cassandra.host/keyspace/table?consistency=ONE&mode=append
parquet:///var/log/?partitionBy=date
elastic://elastic.host/es_index/es_type
mysql://mysql.host/database/table

Parameters:	url (str) – Destination URL.

cassandra(host, keyspace, table, consistency=None, port=None, mode=None, parallelism=None, options=None)[source]¶

Write a dataframe to a Cassandra table.

Parameters:

host (str) – Cassandra server host. keyspace (str) – Cassandra keyspace to write to. table (str) – Cassandra table to write to. consistency (str|None) – Write consistency level: ONE, QUORUM, ALL, etc. port (int|None) – Cassandra server port. mode (str|None) – Spark save mode, http://spark.apache.org/docs/latest/sql-programming-guide.html#save-modes parallelism (int|None) – The max number of parallel tasks that could be executed during the write stage (see Controlling the load). options (dict[str, str]) – Additional options to org.apache.spark.sql.cassandra format (see configuration for Cassandra).

elastic(host, es_index, es_type, port=None, mode=None, parallelism=None, options=None)[source]¶

Write a dataframe into an ElasticSearch index.

Parameters:

host (str) – Elastic server host. es_index (str) – Elastic index. es_type (str) – Elastic type. port (int|None) – mode (str|None) – Spark save mode, http://spark.apache.org/docs/latest/sql-programming-guide.html#save-modes parallelism (int|None) – The max number of parallel tasks that could be executed during the write stage (see Controlling the load). options (dict[str, str]) – Additional options to org.elasticsearch.spark.sql format (see configuration for Elastic).

kafka(host, topic, key_serializer, value_serializer, port=9092, parallelism=None, options=None)[source]¶

Writes dataframe to kafka topic.

The schema of the dataframe should conform the pattern:

>>>  StructType([
...     StructField('key', ...),
...     StructField('value', ...),
...  ])

Parameters key_serializer and value_serializer are callables which get’s python structure as input and should return bytes of encoded data as output.

Parameters:	host (str) – Kafka host. topic (str) – Topic to write to. key_serializer (function) – Function to serialize key. value_serializer (function) – Function to serialize value. port (int) – Kafka port. parallelism (int|None) – The max number of parallel tasks that could be executed during the write stage (see Controlling the load). options (dict|None) – Additional options.

mysql(host, database, table, port=None, mode=None, parallelism=None, options=None)[source]¶

Write a dataframe to a MySQL table.

Options should include user and password.

Parameters:

host (str) – MySQL server address. database (str) – Database to connect to. table (str) – Table to read rows from. mode (str|None) – Spark save mode, http://spark.apache.org/docs/latest/sql-programming-guide.html#save-modes parallelism (int|None) – The max number of parallel tasks that could be executed during the write stage (see Controlling the load). options (dict) – Additional options for JDBC writer (see configuration for MySQL).

sparkly.writer.attach_writer_to_dataframe()[source]¶

A tiny amount of magic to attach write extensions.

About Hive Metastore¶

The Hive Metastore is a database with metadata for Hive tables.

To configure `SparklySession to work with external Hive Metastore, you need to set hive.metastore.uris option. You can do this via hive-site.xml file in spark config ($SPARK_HOME/conf/hive-site.xml):

<property>
  <name>hive.metastore.uris</name>
  <value>thrift://<n.n.n.n>:9083</value>
  <description>IP address (or fully-qualified domain name) and port of the metastore host</description>
</property>

or set it dynamically via SparklySession options:

class MySession(SparklySession):
    options = {
        'hive.metastore.uris': 'thrift://<n.n.n.n>:9083',
    }

Tables management¶

Why: sometimes you need more than just to create a table.

from sparkly import SparklySession


spark = SparklySession()

assert spark.catalog_ext.has_table('my_table') in {True, False}
spark.catalog_ext.rename_table('my_table', 'my_new_table')
spark.catalog_ext.drop_table('my_new_table')

Table properties management¶

Why: sometimes you want to assign custom attributes for your table, e.g. creation time, last update, purpose, data source. The only way to interact with table properties in spark - use raw SQL queries. We implemented a more convenient interface to make your code cleaner.

from sparkly import SparklySession


spark = SparklySession()
spark.catalog_ext.set_table_property('my_table', 'foo', 'bar')
assert spark.catalog_ext.get_table_property('my_table', 'foo') == 'bar'
assert spark.catalog_ext.get_table_properties('my_table') == {'foo': 'bar'}

Note properties are stored as strings. In case if you need other types, consider using a serialisation format, e.g. JSON.

API documentation¶

class sparkly.catalog.SparklyCatalog(spark)[source]¶

A set of tools to interact with HiveMetastore.

drop_table(table_name, checkfirst=True)[source]¶

Drop table from the metastore.

Note

Follow the official documentation to understand DROP TABLE semantic. https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL #LanguageManualDDL-DropTable

Parameters:	table_name (str) – A table name. checkfirst (bool) – Only issue DROPs for tables that are presented in the database.

get_table_properties(table_name)[source]¶

Get table properties from the metastore.

Parameters:	table_name (str) – A table name.
Returns:	Key/value for properties.
Return type:	dict[str,str]

get_table_property(table_name, property_name, to_type=None)[source]¶

Get table property value from the metastore.

Parameters:	table_name (str) – A table name. Might contain a db name. E.g. “my_table” or “default.my_table”. property_name (str) – A property name to read value for. to_type (function) – Cast value to the given type. E.g. int or float.
Returns:	Any

has_table(table_name, db_name=None)[source]¶

Check if table is available in the metastore.

Parameters:	table_name (str) – A table name. db_name (str) – A database name.
Returns:	bool

rename_table(old_table_name, new_table_name)[source]¶

Rename table in the metastore.

Note

Follow the official documentation to understand ALTER TABLE semantic. https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL #LanguageManualDDL-RenameTable

Parameters:	old_table_name (str) – The current table name. new_table_name (str) – An expected table name.

set_table_property(table_name, property_name, value)[source]¶

Set value for table property.

Parameters:	table_name (str) – A table name. property_name (str) – A property name to set value for. value (Any) – Will be automatically casted to string.

Base TestCases¶

There are two main test cases available in Sparkly:

• SparklyTest creates a new session for each test case.
• SparklyGlobalSessionTest uses a single sparkly session for all test cases to boost performance.

from sparkly import SparklySession
from sparkly.test import SparklyTest


class MyTestCase(SparklyTest):
    session = SparklySession

    def test(self):
        df = self.spark.read_ext.by_url(...)

        # Compare all fields
        self.assertDataFrameEqual(
            actual_df=df,
            expected_data=[
                {'col1': 'row1', 'col2': 1},
                {'col1': 'row2', 'col2': 2},
            ],
        )

        # Compare a subset of fields
        self.assertDataFrameEqual(
            actual_df=df,
            expected_data=[
                {'col1': 'row1'},
                {'col1': 'row2'},
            ],
            fields=['col1'],
        )

...

class MyTestWithReusableSession(SparklyGlobalSessionTest):
    context = SparklySession

    def test(self):
        df = self.spark.read_ext.by_url(...)

...

Fixtures¶

“Fixture” is a term borrowed from Django framework. Fixtures load data to a database before the test execution.

There are several storages supported in Sparkly:

• Elastic
• Cassandra (requires cassandra-driver)
• Mysql (requires PyMySql)

_ Kafka (requires kafka-python)

from sparkly.test import MysqlFixture, SparklyTest


class MyTestCase(SparklyTest):
    ...
    fixtures = [
        MysqlFixture('mysql.host',
                     'user',
                     'password',
                     '/path/to/setup_data.sql',
                     '/path/to/remove_data.sql')
    ]
    ...

class sparkly.testing.CassandraFixture(host, setup_file, teardown_file)[source]¶

Fixture to load data into cassandra.

Notes

• Depends on cassandra-driver.

Examples

>>> class MyTestCase(SparklyTest):
...      fixtures = [
...          CassandraFixture(
...              'cassandra.host',
...              absolute_path(__file__, 'resources', 'setup.cql'),
...              absolute_path(__file__, 'resources', 'teardown.cql'),
...          )
...      ]
...

>>> class MyTestCase(SparklyTest):
...      data = CassandraFixture(
...          'cassandra.host',
...          absolute_path(__file__, 'resources', 'setup.cql'),
...          absolute_path(__file__, 'resources', 'teardown.cql'),
...      )
...      def setUp(self):
...          data.setup_data()
...      def tearDown(self):
...          data.teardown_data()
...

>>> def test():
...     fixture = CassandraFixture(...)
...     with fixture:
...        test_stuff()
...

class sparkly.testing.ElasticFixture(host, es_index, es_type, mapping=None, data=None, port=None)[source]¶

Fixture for elastic integration tests.

Examples

>>> class MyTestCase(SparklyTest):
...      fixtures = [
...          ElasticFixture(
...              'elastic.host',
...              'es_index',
...              'es_type',
...              '/path/to/mapping.json',
...              '/path/to/data.json',
...          )
...      ]
...

class sparkly.testing.Fixture[source]¶

Base class for fixtures.

Fixture is a term borrowed from Django tests, it’s data loaded into database for integration testing.

setup_data()[source]¶

Method called to load data into database.

teardown_data()[source]¶

Method called to remove data from database which was loaded by setup_data.

class sparkly.testing.KafkaFixture(host, port=9092, topic=None, key_serializer=None, value_serializer=None, data=None)[source]¶

Fixture for kafka integration tests.

Notes

• depends on kafka-python lib.
• json file should contain array of dicts: [{‘key’: ..., ‘value’: ...}]

Examples

>>> class MyTestCase(SparklySession):
...     fixtures = [
...         KafkaFixture(
...             'kafka.host', 'topic',
...             key_serializer=..., value_serializer=...,
...             data='/path/to/data.json',
...         )
...     ]

class sparkly.testing.KafkaWatcher(spark, df_schema, key_deserializer, value_deserializer, host, topic, port=9092)[source]¶

Context manager that tracks Kafka data published to a topic

Provides access to the new items that were written to a kafka topic by code running within this context.

NOTE: This is mainly useful in integration test cases and may produce unexpected results in production environments, since there are no guarantees about who else may be publishing to a kafka topic.

Usage:

my_deserializer = lambda item: json.loads(item.decode(‘utf-8’)) kafka_watcher = KafkaWatcher(

my_sparkly_session, expected_output_dataframe_schema, my_deserializer, my_deserializer, ‘my.kafkaserver.net’, ‘my_kafka_topic’,

) with kafka_watcher:

# do stuff that publishes messages to ‘my_kafka_topic’

self.assertEqual(kafka_watcher.count, expected_number_of_new_messages) self.assertDataFrameEqual(kafka_watcher.df, expected_df)

class sparkly.testing.MysqlFixture(host, user, password=None, data=None, teardown=None)[source]¶

Fixture for mysql integration tests.

Notes

• depends on PyMySql lib.

Examples

>>> class MyTestCase(SparklyTest):
...      fixtures = [
...          MysqlFixture('mysql.host', 'user', 'password', '/path/to/data.sql')
...      ]
...      def test(self):
...          pass
...

class sparkly.testing.SparklyGlobalSessionTest(methodName='runTest')[source]¶

Base test case that keeps a single instance for the given session class across all tests.

Integration tests are slow, especially when you have to start/stop Spark context for each test case. This class allows you to reuse Spark session across multiple test cases.

class sparkly.testing.SparklyTest(methodName='runTest')[source]¶

Base test for spark scrip tests.

Initialize and shut down Session specified in session attribute.

Example

>>> class MyTestCase(SparklyTest):
...     def test(self):
...         self.assertDataFrameEqual(
...              self.spark.sql('SELECT 1 as one').collect(),
...              [{'one': 1}],
...         )

assertDataFrameEqual(actual_df, expected_data, fields=None, ordered=False)[source]¶

Ensure that DataFrame has the right data inside.

Parameters:	actual_df (pyspark.sql.DataFrame|list[pyspark.sql.Row]) – Dataframe to test data in. expected_data (list[dict]) – Expected dataframe rows defined as dicts. fields (list[str]) – Compare only certain fields. ordered (bool) – Does order of rows matter?

session¶

alias of SparklySession