README.md

spark-local

API enabling switching between Spark execution engine and local implementation based on Scala collections.

Table of contents

• Goals
• Getting started
• Examples
• IO operations
  • Supported formats
  • Data model versioning
  • Column mapping
• Aggregations
• Supported Spark versions
• Supported Spark operations
• Benchmarks
• Contributing

Goals

• Speed up unit testing for Spark applications
• Enable switching between Spark execution engine and Scala collections depending on use case, especially size of data without changing implementation

Getting started

Spark 2.1.1

Include dependency:

"com.github.piotr-kalanski" % "spark-local_2.1.1_2.11" % "0.6.0"

or

<dependency>
    <groupId>com.github.piotr-kalanski</groupId>
    <artifactId>spark-local_2.1.1_2.11</artifactId>
    <version>0.6.0</version>
</dependency>

Spark 2.1.0

Include dependency:

"com.github.piotr-kalanski" % "spark-local_2.1.0_2.11" % "0.6.0"

or

<dependency>
    <groupId>com.github.piotr-kalanski</groupId>
    <artifactId>spark-local_2.1.0_2.11</artifactId>
    <version>0.6.0</version>
</dependency>

Examples

Creating Session

Entry point for library is session object which is similar to SparkSession object from Apache Spark.

Process of creating Session is also similar to Apache Spark.

When creating Session object you can choose between different execution engines. Currently supported:

• Spark - wrapper on Spark, which can be used at production data volumes
• ScalaEager - implementation based on Scala collection with eager transformations, which makes it fast for unit testing
• ScalaLazy - implementation based on Scala collection with lazy transformations, dedicated for working with small/mid size data
• ScalaParallel - implementation based on Scala parallel collection with eager transformations
• ScalaParallelLazy - implementation based on Scala parallel collection with lazy transformations

import com.datawizards.sparklocal.session.ExecutionEngine.ExecutionEngine
import com.datawizards.sparklocal.session.{ExecutionEngine, SparkSessionAPI}

// just change this value to start using different execution engine
val engine = ExecutionEngine.ScalaEager

val session = SparkSessionAPI
      .builder(engine)
      .master("local")
      .getOrCreate()

val ds = session.read[Person](CSVDataStore(file))

RDD API

import com.datawizards.sparklocal.rdd.RDDAPI
import org.apache.spark.sql.SparkSession

object ExampleRDD1 {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local").getOrCreate()

    val data = Seq(1,2,3)
    val rdd = spark.sparkContext.parallelize(data)

    assertEquals(
      calculateSum(RDDAPI(data)),
      calculateSum(RDDAPI(rdd))
    )

    assertEquals(
      calculateSumOfSquares(RDDAPI(data)),
      calculateSumOfSquares(RDDAPI(rdd))
    )

  }

  def assertEquals[T](r1:T, r2:T): Unit = {
    println(r1)
    assert(r1 == r2)
  }

  def calculateSum(ds: RDDAPI[Int]): Int = ds.reduce(_ + _)
  def calculateSumOfSquares(ds: RDDAPI[Int]): Int = ds.map(x=>x*x).reduce(_ + _)

}

Dataset API

Simple example

import com.datawizards.sparklocal.dataset.DataSetAPI
import org.apache.spark.sql.SparkSession

object ExampleDataset1 {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local").getOrCreate()
    import spark.implicits._

    val data = Seq(1,2,3)
    val ds = data.toDS()

    assertEquals(
      calculateSum(DataSetAPI(data)),
      calculateSum(DataSetAPI(ds))
    )

    assertEquals(
      calculateSumOfSquares(DataSetAPI(data)),
      calculateSumOfSquares(DataSetAPI(ds))
    )

  }

  def assertEquals[T](r1:T, r2:T): Unit = {
    println(r1)
    assert(r1 == r2)
  }

  def calculateSum(ds: DataSetAPI[Int]): Int = ds.reduce(_ + _)
  def calculateSumOfSquares(ds: DataSetAPI[Int]): Int = ds.map(x=>x*x).reduce(_ + _)

}

Example report

case class Person(id: Int, name: String, gender: String)
case class WorkExperience(personId: Int, year: Int, title: String)
case class HRReport(year: Int, title: String, gender: String, count: Int)

object ExampleHRReport {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local").getOrCreate()
    import spark.implicits._

    val people = SampleData.people
    val peopleDs = people.toDS()
    val experience = SampleData.experience
    val experienceDs = experience.toDS()


    calculateReport(DataSetAPI(people), DataSetAPI(experience))
    calculateReport(DataSetAPI(peopleDs), DataSetAPI(experienceDs))
  }

  def calculateReport(people: DataSetAPI[Person], workExperience: DataSetAPI[WorkExperience]): DataSetAPI[HRReport] = {
    workExperience
      .join(people)(_.personId, _.id)
      .groupByKey(wp => (wp._1.year, wp._1.title, wp._2.gender))
      .mapGroups{case ((year, title, gender), vals) => HRReport(year, title, gender, vals.size)}
  }

}

IO operations

Library provides dedicated API for input/output operations with implementation for Spark and Scala collections.

Supported formats

Supported formats:

• CSV
• JSON
• Parquet
• Avro
• Hive
• JDBC
  • H2
  • MySQL
• Elasticsearch

CSV

Read CSV file

val reader: Reader = ReaderScalaImpl // Scala implementation
//val reader: Reader = ReaderSparkImpl // Spark implementation

reader.read[Person](
    CSVDataStore(
        path = "people.csv",
        delimiter = ';',
        header = false,
        columns = Seq("name","age")
    )
)

Write to CSV file

ds.write(CSVDataStore(file), SaveMode.Overwrite)

JSON

Read JSON file

reader.read[Person](JsonDataStore("people.json"))

Write to JSON file

ds.write(JsonDataStore("people.json"), SaveMode.Overwrite)

Avro

Current implementation produces different binary files for Spark and Scala. Spark by default compress files with snappy and spark-local implementation is based on: https://github.com/sksamuel/avro4s, which saves data without compression.

Read Avro file

reader.read[Person](AvroDataStore("people.avro"))

Write to Avro file

ds.write(AvroDataStore("people.avro"), SaveMode.Overwrite)

Parquet

Read Parquet file

reader.read[Person](ParquetDataStore("people.parquet"))

Write to Parquet file

ds.write(ParquetDataStore("people.parquet"), SaveMode.Overwrite)

Hive

Read Hive table

reader.read[Person](HiveDataStore("db", "table"))

Write to Hive table

ds.write(HiveDataStore("db", "table"), SaveMode.Overwrite)

JDBC

Read JDBC table

val database = "public"
val table = "people"
val properties = new java.util.Properties()
reader.read[Person](H2DataStore(connectionString, database, table, properties))

Write to JDBC table

ds.write(H2DataStore(connectionString, database, table, properties), SaveMode.Append)

Elasticsearch

Write to Elasticsearch index

val indexName = "people"
val typeName = "person"
ds.write(ElasticsearchSimpleIndexDataStore("localhost", indexName, typeName), SaveMode.Append)

Data model versioning

Library supports reading data using old and new version of data model.

Reading with old version of data model is straightforward, just existing fields will be read. On other hand, when reading with new version of data model, new columns should be Option type and value None is assigned to those fields.

Example

import com.datawizards.sparklocal.datastore.CSVDataStore
import com.datawizards.sparklocal.session.{ExecutionEngine, SparkSessionAPI}
import org.apache.spark.sql.SaveMode

case class Person(name: String, age: Int)
case class PersonV2(name: String, age: Int, title: Option[String])
case class PersonV3(name: String, age: Int, title: Option[String], salary: Option[Long])

val session = SparkSessionAPI
      .builder(ExecutionEngine.ScalaEager)
      .master("local")
      .getOrCreate()

import session.implicits._

val peopleV2 = session.createDataset(Seq(
    PersonV2("p1", 10, Some("Mr")),
    PersonV2("p2", 20, Some("Ms")),
    PersonV2("p3", 30, None),
    PersonV2("p4", 40, Some("Mr"))
))

val dataStore = CSVDataStore("people_v2.csv")
peopleV2.write(dataStore, SaveMode.Overwrite)

// Read old version:
val peopleV1 = session.read[Person](dataStore)

// Read new version:
val peopleV3 = session.read[PersonV3](dataStore)

peopleV1.show()
peopleV3.show()

+----+---+
|name|age|
+----+---+
|p1  |10 |
|p2  |20 |
|p3  |30 |
|p4  |40 |
+----+---+

+----+---+-----+------+
|name|age|title|salary|
+----+---+-----+------+
|p1  |10 |Mr   |      |
|p2  |20 |Ms   |      |
|p3  |30 |     |      |
|p4  |40 |Mr   |      |
+----+---+-----+------+

Column mapping

Library supports changing name of fields when writing and reading data.

Mapping is provided using column annotation from project: https://github.com/piotr-kalanski/data-model-generator.

Example

import com.datawizards.dmg.annotations.column
import com.datawizards.sparklocal.dataset.io.ModelDialects
import com.datawizards.sparklocal.datastore.JsonDataStore
import com.datawizards.sparklocal.session.{ExecutionEngine, SparkSessionAPI}
import org.apache.spark.sql.SaveMode

case class PersonWithMapping(
    @column("personName", dialect = ModelDialects.JSON)
    name: String,
    @column("personAge", dialect = ModelDialects.JSON)
    age: Int
)

val session = SparkSessionAPI
      .builder(ExecutionEngine.ScalaEager)
      .master("local")
      .getOrCreate()

import session.implicits._

val people = session.createDataset(Seq(
    PersonWithMapping("p1", 10),
    PersonWithMapping("p2", 20),
    PersonWithMapping("p3", 30),
    PersonWithMapping("p4", 40)
))

val dataStore = JsonDataStore("people_mapping.json")
people.write(dataStore, SaveMode.Overwrite)

Name of fields in JSON file are consistent with column names provided in annotations:

{"personName":"p1","personAge":10}
{"personName":"p2","personAge":20}
{"personName":"p3","personAge":30}
{"personName":"p4","personAge":40}

Reading data:

val people2 = session.read[PersonWithMapping](dataStore)
people2.show()

Names of columns are the same as case class fields:

+----+---+
|name|age|
+----+---+
|p1  |10 |
|p2  |20 |
|p3  |30 |
|p4  |40 |
+----+---+

Aggregations

Library provides custom type-safe API for aggregations.

Example operations:

import com.datawizards.sparklocal.dataset.agg._

ds
    .groupByKey(_.name)
    .agg(sum(_.age), count(), mean(_.age), max(_.age))

Supported Spark versions

spark-local	Spark version
0.7	2.1.1 2.1.0
0.6	2.1.1 2.1.0
0.5	2.1.0
0.4	2.1.0
0.3	2.1.0
0.2	2.1.0
0.1	2.1.0

Bugs

Please report any bugs to spark-local Github issue tracker.

Continuous Integration

Build History

Contact

piotr.kalanski@gmail.com