About Sparkly Context¶

SparklyContext class is the main class of the Sparkly library. It encompasses all of this library’s functionality. Most of times you want to subclass it to define the various options you desire through class attributes.

Sparkly context have links to other extras of the lib:

Dataframe pyspark class is also monkey patched with write_ext (Read/write utilities for DataFrames) attribute for convenient writing.

Use cases¶

Cassandra¶

Sparkly relies on the official spark cassandra connector and was successfully tested in production using versions 1.5.x and 1.6.x.

from sparkly import SparklyContext


class MyContext(SparklyContext):
    # Feel free to play with other versions
    packages = ['datastax:spark-cassandra-connector:1.6.1-s_2.10']

hc = MyContext()

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

CSV¶

Sparkly relies on the csv connector provided by Databricks.

from sparkly import SparklyContext


class MyContext(SparklyContext):
    # Feel free to play with other versions
    packages = ['com.databricks:spark-csv_2.10:1.4.0']

hc = MyContext()

# To read data
df = hc.read_ext.csv('/path/to/csv/file.csv', header=True)
# To write data
df.write_ext.csv('/path/to/csv/file.csv', header=False)

Elastic¶

Sparkly relies on the official elastic spark connector and was successfully tested in production using versions 2.2.x and 2.3.x.

from sparkly import SparklyContext


class MyContext(SparklyContext):
    # Feel free to play with other versions
    packages = ['org.elasticsearch:elasticsearch-spark_2.10:2.3.0']

hc = MyContext()

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

MySQL¶

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

from sparkly import SparklyContext
from sparkly.utils import absolute_path


class MyContext(SparklyContext):
    # Feel free to play with other versions.
    jars = [absolute_path(__file__, './path/to/mysql-connector-java-5.1.39-bin.jar')]

hc = MyContext()

# To read data
df = hc.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
})

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 SparklyContext

class MyContext(SparklyContext):
    packages = [
        'org.apache.spark:spark-streaming-kafka_2.10:1.6.1',
    ]

hc = MyContext()

# To read data from kafka in json as Dataframe.

#   1. Define the schema of the data 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 the 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 data to kafka in json from Dataframe
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'),
)

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(hc)[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

csv(path, custom_schema=None, header=True, parallelism=None, options=None)[source]¶

Create a dataframe from a CSV file.

Parameters:	path (str) – Path to the file or directory. custom_schema (pyspark.sql.types.DataType) – Force custom schema. header (bool) – The first row is a header. 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 com.databricks.spark.csv format. (see configuration for CSV).
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’s 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.

Should be usable for rds, aurora, etc. 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 the 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).

csv(path, header=False, mode=None, partitionBy=None, parallelism=None, options=None)[source]¶

Write a dataframe to a CSV file.

Parameters:

path (str) – Path to the output directory. header (bool) – First row is a header. mode (str|None) – Spark save mode, http://spark.apache.org/docs/latest/sql-programming-guide.html#save-modes partitionBy (list[str]) – Names of partitioning columns. 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 com.databricks.spark.csv format (see configuration for CSV).

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.

Should be usable for rds, aurora, etc. 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¶

Hive metastore is a database storing metadata about Hive tables, which you operate in your Sparkly (Hive) Context. Read more about Hive Metastore

To configure a SparklyContext to work with your Hive Metastore, you have to set hive.metastore.uris option. You can do this via hive-site.xml file in spark config ($SPARK_HOME/conf/hive-site.xml), like this:

<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 in SparklyContext options, like this:

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

After this your sparkly context will operate on the configured Hive Metastore.

Use cases¶

from sparkly import SparklyContext
hc = SparklyContext()
assert(hc.hms.table('my_table').exists() in {True, False})

from sparkly import SparkeContext
# input
hc = SparklyContext()
df = hc.read_ext.by_url('parquet:s3://path/to/data/')
# operation
hc.hms.create_table(
     'new_shiny_table',
     df,
     location='s3://path/to/data/',
     partition_by=['partition', 'fields'],
     output_format='parquet'
)
new_df = hc.read_ext.by_url('table://new_shiny_table')

from sparkly import SparkeContext
# input
hc = SparklyContext()
df = hc.read_ext.by_url('csv:s3://path/to/data/new/')
# operation
table = hc.hms.replace_table(
    'old_table',
    df,
    location='s3://path/to/data/',
    partition_by=['partition', 'fields'],
)

from sparkly import SparklyContext
hc = SparklyContext()
table = hc.hms.table('my_table')
table.set_property('foo', 'bar')
assert table.get_property('foo') == 'bar'
assert table.get_all_properties() == {'foo': 'bar'}

API documentation¶

class sparkly.hive_metastore_manager.SparklyHiveMetastoreManager(hc)[source]¶

A set of tools to interact with HiveMetastore.

create_table(table_name, schema, location, partition_by=None, table_format=None, properties=None)[source]¶

Creates table in Hive Metastore.

Parameters:	table_name (str) – name of new Table. schema (pyspark.sql.dataframe.DataFrame) – schema. location (str) – location of data. partition_by (list) – partitioning columns. table_format (str) – default is parquet. properties (dict) – properties to assign to the table.
Returns:	Table

get_all_tables()[source]¶

Returns all tables available in metastore.

Returns:	list

replace_table(table_name, schema, location, partition_by=None, table_format=None)[source]¶

Replaces table table_name with data represented by schema, location.

Parameters:	table_name (str) – Table name. schema (pyspark.sql.dataframe.DataFrame) – schema. location (str) – data location, ex.: s3://path/tp/data. partition_by (list) – fields the data partitioned by.
Returns:	Table

class sparkly.hive_metastore_manager.Table(hms, table_name)[source]¶

Represents a table in HiveMetastore.

Provides meta data operations on a Table.

df()[source]¶

Returns dataframe for the managed table.

Returns:	pyspark.sql.dataframe.DataFrame

exists()[source]¶

Checks if table exists.

Returns:	bool

get_all_properties()[source]¶

Returns all table properties.

Returns:	Property names to values.
Return type:	dict

get_property(name, to_type=None)[source]¶

Gets table property.

Parameters:	name (str) – Name of the property. to_type (type) – Type to coarce to, str by default.
Returns:	any

set_property(name, value)[source]¶

Sets table property.

Parameters:	name (str) – Name of the property. value (str) – Value of the proporty.
Returns:	Self.
Return type:	Table

Use cases¶

from sparkly.schema_parser import generate_structure_type, parse_schema

data = ... parse data from file ...
schema_as_string = 'name:string|age:int'  # Note: you can get this from command line, for example
spark_schema = generate_structure_type(parse_schema(schema_as_string))
df = ctx.createDataframe(data, spark_schema)

Base testing classes¶

There are two main testing classes in Sparkly:

• SparklyTest:

  • Instantiates Sparkly context specified in context attribute.
  • The context will be available via self.hc.

• SparklyGlobalContextTest:

  • Reuses single SparklyContext for all tests for performance boost.

Example:

from sparkly import SparklyContext
from sparkly.test import SparklyTest

class MyTestCase(SparklyTest):
    context = SparklyContext
    def test(self):
        df = self.hc.read_ext.by_url(...)
        self.assertDataFrameEqual(
            df, [('test_data', 1)], ['name', 'number']
        )

...

class MyTestWithReusableContext(SparklyGlobalContextTest):
    context = SparklyContext
    def test(self):
        df = self.hc.read_ext.by_url(...)

...

Fixtures¶

Fixtures is term borrowed from testing in Django framework. A fixture will load data to a database upon text execution.

There are couple of databases supported in Sparkly:

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

Example:

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.

Notes

• Data upload uses bulk api.

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(SparklyContext):
...     fixtures = [
...         KafkaFixture(
...             'kafka.host', 'topic',
...             key_serializer=..., value_serializer=...,
...             data='/path/to/data.json',
...         )
...     ]

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.SparklyGlobalContextTest(methodName='runTest')[source]¶

Base test case that keeps a single instance for the given context 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 context across multiple test cases.

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

Base test for spark scrip tests.

Initializes and shuts down Context specified in context param.

Example

>>> class MyTestCase(SparklyTest):
...     def test(self):
...         self.assertDataFrameEqual(
...              self.hc.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?

context¶

alias of SparklyContext