A demo script for loading data into Spark and performing a few basic analyzes (from presentation http://4quant.github.io/spark-introduction on slide https://rawgit.com/4Quant/spark-introduction/master/TutorialSlides.html#/26). To get started make a clone of the repository locally (git clone https://gist.github.com/755c2d99c23f4cbe2e74.git) and p…

SparkDemo.scala

val inData=sc.textFile("tiny.csv")
// Read the first line and split by commas
import scala.collection.immutable.StringOps
val header= inData.first
val headerLine = header.split(",").map(_.trim.filter(_ != '"'))
// Count lines in the file
inData.count
// Remove the header from the data
val rowData = inData.filter(_ != header).cache
// Divide each row into columns by seperating by commas
val colData = rowData.map(_.trim.split(",").map(_.toDouble))
// Transform the data set into a Key-Value dataset
import org.apache.spark.SparkContext._
import org.apache.spark.rdd.PairRDDFunctions._
// Define a basic format (case class) and register as a table
@serializable case class CellShape(sample: Long,PosX: Double, PosY: Double, PosZ: Double,Volume: Double) 
val classedData = colData.map(line => CellShape(line(0).toLong,line(5),line(6),line(7),line(31))).cache
/**
 * Find the nearest neighbors
 *
 **/
import scala.math._
// Cross the two datasets
val crossData = classedData.cartesian(classedData).filter{dPair => dPair._1.sample != dPair._2.sample}
// Calculate the Euclidean distance between two points
val distData = crossData.map{dPair => 
   (dPair._1.sample,
    (sqrt(pow(dPair._1.PosX-dPair._2.PosX,2)+pow(dPair._1.PosY-dPair._2.PosY,2)+pow(dPair._1.PosZ-dPair._2.PosZ,2)),
     dPair._1,dPair._2)
    )
}
// Group the data by point
val pointData = distData.groupByKey.map{cvals => 
     val minDist = cvals._2.map(_._1).min
     (cvals._1,cvals._2.filter(ival => ival._1==minDist).head)
}
// Calculate the nearest neighbor distance
val avgNNdist = pointData.map(_._2._1).reduce(_ + _)/pointData.count
/** 
 * Using Spark SQL for analysis
 * Import the basic libraries
**/
import org.apache.spark.sql.SQLContext
val sqlContext = new SQLContext(sc)
import sqlContext._
classedData.registerAsTable("Summary")
// Performing a simple query
sql("SELECT SUM(PosX) from Summary").collect

/**
 * Automatically detecting format from CSV file
 * based on Spark pull request 1351 (https://github.com/apache/spark/pull/1351/)
 **/
import org.apache.spark.sql.catalyst.types._
import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
import org.apache.spark.sql.catalyst.expressions.{GenericMutableRow, AttributeReference, Row}
import org.apache.spark.sql.SchemaRDD
val fields = headerLine.map( fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)
def castToType(value: Any, dataType: DataType): Any = dataType match {
  case StringType => value.asInstanceOf[String]
  case BooleanType => value.asInstanceOf[Boolean]
  case DoubleType => value.asInstanceOf[Double]
  case FloatType => value.asInstanceOf[Float]
  case IntegerType => value.asInstanceOf[Int]
  case LongType => value.asInstanceOf[Long]
  case ShortType => value.asInstanceOf[Short]
  case _ => null
}
def parseCSVLine(splitLine: Array[String], schema: StructType): Row = {
  val row = new GenericMutableRow(schema.fields.length)
  schema.fields.zipWithIndex.foreach {
    case (StructField(name, dataType, _), index) =>
    row.update(index, castToType(splitLine(index), dataType))
  }
  row
}

val asAttr = schema.fields.map(field => AttributeReference(field.name, field.dataType, nullable = true)())
val parsedCSV = rowData.map(_.split(",")).map(line => parseCSVLine(line,schema))
val sqlRdd = new SchemaRDD(sqlContext,SparkLogicalPlan(ExistingRdd(asAttr, parsedCSV)))
/**
 * Running a few SQL commands
 * 
 **/
sqlRdd.registerAsTable("FullTable")

sql("""SELECT a.LACUNA_NUMBER,COUNT(*),AVG(a.POS_X-b.POS_X),AVG(a.POS_Y-b.POS_Y),AVG(a.POS_Z-b.POS_Z) FROM FullTable a 
 JOIN FullTable b ON a.LACUNA_NUMBER!=b.LACUNA_NUMBER AND a.POS_X>b.POS_X AND a.POS_Y>b.POS_Y AND a.POS_Z>b.POS_Z GROUP BY a.LACUNA_NUMBER """).first


/** 
 * Using MLLib with Spark SQL
 * 
 **/
import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.linalg.Vectors

val sqlResults = sql("SELECT POS_X*SCALE_X,POS_Y*SCALE_Y,POS_Z*SCALE_Z FROM FullTable").map{pts => pts.map(_.toString.toDouble)}
val kModel = KMeans.train(sqlResults.map{pts => Vectors.dense(pts.toArray)},4,1000)
kModel.clusterCenters
kModel.predict(Vectors.dense(0,0,0))
kModel.predict(Vectors.dense(1,1,1))

tiny.csv