My Journey in Spark, Part 1: Basic

RDD - Resilient Distributed Datasets

immutable, partitioned, Lazy evaluation, fault-tolerant

transformation vs action

DataSet - static typed

DataFrame = DataSet[Row]

df.as(Encoders.bean(classOf[ModelClass]);

PairRDDFunctions, OrderedRDDFunctions and GroupedRDDFunctions

SparkSession.builder().master("local[*]")

local uses 1 thread.
local[N] uses N threads.
local[*] uses as many threads as there are cores.
local[N, M] and local[*, M]
DAGScheduler, TaskScheduler
Driver, Worker
- Understand what code is executing in worker or in worker.
Datasets, Dataframe
- type-safe, object-oriented programming interface

Read Compressed File
Spark can read compressed file as long as it ends with gz(created by gzip), not tar.gz - link

When read gz file, spark will only give one RDD as gzipped file is not splittable. You may need repartition the RDD. 

Understand what's executed on the Driver vs. Workers

Don't collect huge data in driver

Serialization Errors

- Create the Non-Serializable object in the worker

- Use foreachPartition, create only one Non-Serializable object per partition.

Prefer reduceByKey, aggregateByKey, foldByKey, and combineByKey over groupByKey

-- GroupByKey doesn't do combine in each node

Logging in spark executor code
import org.slf4j.LoggerFactory
@transient lazy val logger = LoggerFactory.getLogger(getClass)

Passing Functions to Spark

cache vs persist
cache = persit(MEMORY_ONLY)

use StorageLevel.MEMORY_AND_DISK if there is no engough memory

Spark-CSV
Check CSVOptions.scala for all options supported by spark-csv.
To deal double quotes in csv: change "escape" from default \ to "

val csvInput = spark.read
  .option("header", "true")
  .option("inferschema", true)
  .option("escape", "\"")
  .option("ignoreLeadingWhiteSpace", true)
  .option("ignoreTrailingWhiteSpace ", true)
  .csv(csvFile)

WebUI
http://localhost:4040/jobs/
Using Either to handle good/bad data

Concepts
Driver, Executors, Cluster Managers

Wide Dependencies
map, filter, mapPartitions and flatMap, coalesce
Narrow Dependencies
sort, reduceByKey, groupByKey, join, and anything that calls for repartition/shuffle

A job is defined an action, wide transformations break jobs into stages.

Spark ETL: Using Either to handle invalid data

Senario
Usually when we use spark to import data from external system, we want to report to client how many rows we have imported successfully, how many invalid rows in the origin input.

Solution: Using Scala Either
When we use map to parse the data, we can use Scala Either: left will contain the invalid data and its error message, right will contain valid data which will be processed and stored later.

SparkService
Here we use dataSet.rdd.map not dataSet.map.

If we tried to call dataSet.map and use Either, it will fail with exception:
java.lang.NoClassDefFoundError: no Java class corresponding to Product with Serializable with scala.util.Either[(String, String),xx.EventModel] found

This seems because in current latest scala http://www.scala-lang.org/api/2.11.x/index.html#scala.util.Either, Either doesn't implements:

Serializable traitsealed abstract class Either[+A, +B] extends AnyRef

In future 2.12 http://www.scala-lang.org/api/2.12.x/scala/util/Either.html, it does:
sealed abstract class Either[+A, +B] extends Product with Serializable

In order to use java 8 types such as Optional, java.time in rest api, use jackson-datatype-jdk8 and jackson-datatype-jsr310.

To use scala case class with Jackson: add @BeanProperty to the field.
Check jackson related tips.
Please check Using Lombok to Simplify the Code for SpringContextBridge implementation.

@Service
@transient lazy val logger = LoggerFactory.getLogger(getClass)
def save(csvFile: String): DataImportResult = {
  val startDate = ZonedDateTime.now
  val spark = SpringContextBridge.getBean(classOf[SparkSession])
  val csvInput = spark.read
    .option("header", "true")
    .option("inferschema", false)
    .option("ignoreLeadingWhiteSpace", true)
    .option("ignoreTrailingWhiteSpace ", true)
    .csv(csvFile)

  val newRdd = csvInput.rdd
    .map { row =>
      try {
        val event = new EventModel()
        event.setId(row.getString(0))
        event.setEventDate(Util.ParseDate(row.getString(1)))
        // ...
        event.setUpdateDate(new Date())
        Right(event)
      } catch {
        case e: Throwable => Left(row.toSeq.map({ _.toString() }).toString(), e.getMessage)
      }
    }
  newRdd.cache()
  val failedRdd = newRdd.map(_.left).filter(_.e.isLeft).map(_.get)

  val failedCount = failedRdd.count()
  val errorData = failedRdd.take(10)
  val successRdd = newRdd.map(_.right).filter(_.e.isRight).map(_.get);
  successRdd.cache
  successRdd //... other process
  .foreachPartition { it =>
   {
     val repo = SpringContextBridge.getBean(classOf[EventRepo])
     it.grouped(1000).foreach { x => repo.saveWithoutCommit(x.toIterable.asJava) }
   }
  }

  SpringContextBridge.getBean(classOf[EventRepo]).hardCommit()

  val validDataCount = successRdd.count
  val result = DataImportResult(startDate, ZonedDateTime.now, startDate.until(ZonedDateTime.now, ChronoUnit.MILLIS), validDataCount, failedCount, errorData)
  logger.info(result.toString())
  result
}
case class DataImportResult(
  @BeanProperty startTime: ZonedDateTime, @BeanProperty endTime: ZonedDateTime, @BeanProperty timeTakenMill: Long,
  @BeanProperty validCount: Long, @BeanProperty failedCount: Long, @BeanProperty errorData: Array[(String, String)])

object MyEncoders {
  implicit def eventEncoder: = org.apache.spark.sql.Encoders.bean[classOf[EventModel]]
}

SparkConfiguration

@Configuration
class SparkConfiguration {
  val logger = LoggerFactory.getLogger(getClass)

  @Bean
  def sparkSessionConfiguration = {
    val spark = SparkSession
      .builder().master("spark://master:port")
      .appName("My Spark App")
      .config("some.config", "some.value")
      .getOrCreate()
    logger.info(s"Created SparkSession: ${spark}")
    spark
  }
}

Maven pom.xml
Add dependencies: spark-core_2.11, spark-sql_2.11 and org.psnively:spring_scala_3-2-14_2.11
To use mixed scala and java, add net.alchim31.maven:scala-maven-plugin.

Spark Basic Statistics - Using Scala

Summary statistics
colStats() returns an instance of MultivariateStatisticalSummary, which contains the column-wise max, min, mean, variance, and number of nonzeros, as well as the total count.
Test data:
1 2 3
10 20 30
100 200 300

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
  
val data = sc.textFile("E:/jeffery/src/ML/data/statistics.txt").cache();  
val parsedData = data.map( line =>  Vectors.dense(line.split(' ').map(x => x.toDouble).toArray) )
val summary = Statistics.colStats(parsedData);
println(summary.count)
println(summary.min)
println(summary.max)
println(summary.mean) // a dense vector containing the mean value for each column
println(summary.variance) // column-wise variance
println(summary.numNonzeros) // number of nonzeros in each column

Stratified sampling
Stratified sampling methods, sampleByKey and sampleByKeyExact, can be performed on RDD’s of key-value pairs.

The sampleByKey method will flip a coin to decide whether an observation will be sampled or not, therefore requires one pass over the data, and provides an expected sample size. sampleByKeyExact requires significant more resources than the per-stratum simple random sampling used in sampleByKey, but will provide the exact sampling size with 99.99% confidence.

Test Dataman 6
woman 14
woman 19
child 6
baby 1
child 3
woman 26

import org.apache.spark.SparkContext._
import org.apache.spark.rdd.PairRDDFunctions
val data = sc.textFile("E:/jeffery/src/ML/data/sampling.txt").cache();  
val parsedData = data.map{line => {
  val sp = line.split(' '); 
  (sp(0), sp(1).toInt);
}
}.cache()

parsedData.foreach(println)
var fractions = Map[String, Double]()

fractions += ("man" ->  0.5, "woman" -> 0.5, "child" -> 0.5, "baby" -> 0.3);
val approxSample = parsedData.sampleByKey(false, fractions).collect();
val exactSample = parsedData.sampleByKeyExact(false, fractions).collect();
print(approxSample.mkString(" "));
print(exactSample.mkString(" "));

Random data generation

import org.apache.spark.mllib.random.RandomRDDs._
val u = normalRDD(sc, 100L, 2);
// Apply a transform to get a random double RDD following `N(1, 4)`.
val v = u.map(x => 1.0 + 2.0 * x)
print(u.collect())
print(v.collect())

val u = poissonRDD(sc, 10, 100L);
val v = u.map(x => 1.0 + 2.0 * x).collect()

val u = uniformRDD(sc, 100L);
val v = u.map(x => 1.0 + 2.0 * x).collect()

Histogram

val ints = sc.parallelize(1 to 100)
ints.histogram(5) // 5 evenly spaced buckets

res92: (Array[Double], Array[Long]) = (Array(1.0, 20.8, 40.6, 60.4, 80.2, 100.0),Array(20, 20, 20, 20, 20)) Correlations


MLlib - Basic Statistics
Spark 1.1.0 Basic Statistics（上）

Build Spark Failure: Nonzero exit code (128): git clone sbt-pom-reader.git

The Problem
Download Sprak 1.2 from github, and try to build it by running sbt assembly.
It always failed with error:
[error] Nonzero exit code (128): git clone https://github.com/ScrapCodes/sbt-pom-reader.git C:\Users\jyuan\.sbt\0.13\staging\ad8e8574a5bcb2d22d23\sbt-pom-reader
[error] Use 'last' for the full log.
Project loading failed: (r)etry, (q)uit, (l)ast, or (i)gnore?

Retry didn't work, and I can access https://github.com/ScrapCodes/sbt-pom-reader.git, git clone it.
Not sure why it failed.

The Solution
To fix this: I opened a new cmd terminal, and ran the following command to create the staging folder and git clone to the dest folder:
mkdir C:\Users\jyuan\.sbt\0.13\staging\ad8e8574a5bcb2d22d23\sbt-pom-reader
git clone https://github.com/ScrapCodes/sbt-pom-reader.git C:\Users\jyuan\.sbt\0.13\staging\ad8e8574a5bcb2d22d23\sbt-pom-reader

Then I type r to retry it. As the sbt-pom-reader is already there, sbt would just happily take it. 
After several minutes, spark built succesfully

Happy hacking.