Introduction to Apache Spark

 mpirun -np 4 simple1

  // determining who am I
  int myrank;
  MPI_Comm_rank(MPI_COMM_WORLD, &myrank); (1)
  // determining who are they :
  int nprocs;
  MPI_Comm_size(MPI_COMM_WORLD, &nprocs);  (2)
  // sending messages
  MPI_Send(buffer, count, datatype, destination, tag, MPI_COMM_WORLD); (3)
  //receiving Messages
  MPI_Recv(buffer, maxcount, datatype, source, tag, MPI_COMM_WORLD, &status); (4)

val data = 1 to 1000
val filteredData= data.filter( _%2==0)

val data = 1 to 1000
val filteredData = sc.parallelize(data).filter(_%2==0) (1) (2)
filteredData.collect (3)

val data = 1 to 1000 (1)
val firstRDD = sc.parallelize(data) (2)
val secondRDD = firstRDD.filter( _ < 10) (3)
secondRDD.collect (4)

val data = 1 to 1000 (1)

val firstRDD = sc.parallelize(data) (1)

def parallelize[T](seq: Seq[T], numSlices: Int): rdd.RDD[T]

val secondRDD = firstRDD.filter( _ < 10)

secondRDD.collect (4)

val input = sc.textFile("s3://...") (1)
val words = input.flatMap(x => x.split(" ")) (2)
val result = words.map(x => (x, 1))(3)
                  .reduceByKey((x, y) => x + y) (4)

val input = sc.textFile("s3://...") (1)
val result = lines.flatMap(x => x.split(" ")).countByValue() (2)

val data = 1 to 1000   (1)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)
val groupedData = sortedData.groupBy( (x:Int)=> x%3) (5)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)
val groupedData = sortedData.groupBy( (x:Int)=> x%3) (5)
groupedData.collect (6)

val sqlContext = new org.apache.spark.sql.SQLContext(sc) (1)

case class Person(name: String, age: Int) (2)

val people = sc.textFile("persons.txt")
            .map(_.split(","))
            .map(p => Person(p(0), p(1).trim.toInt)
            .toDF (3)

people.createOrReplaceTempView("people") (4)

val teenagers = sql("""
        SELECT * from people
        WHERE age < 15
        """) (5)

teenagers.show (6)

  CassandraConnector(conf).withSessionDo { session =>
    session.execute(s"""
        CREATE KEYSPACE IF NOT EXISTS demo
        WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }""")
    session.execute(s"CREATE TABLE IF NOT EXISTS demo.wordcount (word TEXT PRIMARY KEY, count COUNTER)")
  }

  sc.textFile(words)
    .flatMap(_.split("\\s+"))
    .map(word => (word.toLowerCase, 1))
    .reduceByKey(_ + _)
    .saveToCassandra("demo", "wordcount")

  // print out the data saved from Spark to Cassandra
  sc.cassandraTable("demo", "wordcount")
    .collect.foreach(println)

val data = 1 to 1000   (1)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)
val groupedData = sortedData.groupBy( (x:Int)=> x%3) (5)

val data = 1 to 1000   (1)
val parallelizedData = sc.parallelize(data)  (2)
val filteredData = parallelizedData.filter(_%2==0)  (3)
val sortedData = filteredData.sortBy( (x:Int) => - x ) (4)
val groupedData = sortedData.groupBy( (x:Int)=> x%3) (5)
groupedData.collect (6)

  CassandraConnector(conf).withSessionDo { session =>
    session.execute(s"""
        CREATE KEYSPACE IF NOT EXISTS demo
        WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }""")
    session.execute(s"CREATE TABLE IF NOT EXISTS demo.wordcount (word TEXT PRIMARY KEY, count COUNTER)")
  }

  sc.textFile(words)
    .flatMap(_.split("\\s+"))
    .map(word => (word.toLowerCase, 1))
    .reduceByKey(_ + _)
    .saveToCassandra("demo", "wordcount")

  // print out the data saved from Spark to Cassandra
  sc.cassandraTable("demo", "wordcount")
    .collect.foreach(println)

val words = Array("one", "two", "two", "three", "three", "three")
val wordPairsRDD = sc.parallelize(words).map(word => (word, 1))

val wordCountsWithReduce = wordPairsRDD
  .reduceByKey(_ + _)
  .collect()

val wordCountsWithGroup = wordPairsRDD
  .groupByKey()
  .map(t => (t._1, t._2.sum))
  .collect()

var counter = 0
var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach(x => counter += x)

println("Counter value: " + counter)

./bin/spark-shell --master local[*]

./bin/spark-shell --master spark://HOST1:PORT1
./bin/spark-shell --master spark://HOST1:PORT1,HOST2:PORT2

./bin/spark-shell --master mesos://HOST:PORT
./bin/spark-shell --master yarn
./bin/spark-shell --master k8s://HOST:PORT

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)

var myClass = new NonSerializableClass()

var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach( ... myClass ...)

var myClass = new NonSerializableClass()

var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach( ... myClass ...)

object AwsClient{
    val s3 = new AmazonS3Client(new BasicAWSCredentials(AWS_ID, AWS_KEY))
}


sampleDF.foreach( r=> {
            ...
            AwsClient.s3.putObject(... ,... ,... )
})

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
scala> accum.value
res2: Long = 10

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

val myRDD = sc.parallelize(persons)
myRDD.map(x=>(x.dept,(x.age,1)))
     .reduceByKey((a,b)=>(a._1+b._1,a._2+b._2))
     .map(p=>(p._1,p._2._1/p._2._2))
     .collect

val myRDD = sc.parallelize(persons)
myRDD.map(x=>(x.dept,(x.age,1)))
     .reduceByKey((a,b)=>(a._1+b._1,a._2+b._2))
     .map(p=>(p._1,p._2._1/p._2._2))
     .collect

 --- SQL LIKE
 SELECT avg(age) FROM pdata
 GROUP BY dept;

val myRDD = sc.parallelize(persons)
myRDD.map(x=>(x.dept,(x.age,1)))
     .reduceByKey((a,b)=>(a._1+b._1,a._2+b._2))
     .map(p=>(p._1,p._2._1/p._2._2))
     .collect

data.groupBy("dept").avg("age")

// Create from RDD
val dfFromRDD = myRDD.df

// Create from JSON
val df = sparkSession.read.json("examples/src/main/resources/people.json")

// Show the content of the DataFrame
df.show()
// age  name
// 30   Andy
// 19   Justin

// Print the schema in a tree format
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// Select only the "name" column
df.select("name").show() // implicit conversion from string to Column
// name
// Michael
// Andy
// Justin

//Equivalent column selection
df.select(df("name")).show() // apply function on dataframe

df.select(col("name")).show() //using the col function

df.select(df.col("name")).show() //using the col method

df.select($"name").show // easy scala shorcut

// Select everybody, but increment the age by 1
df.select(df("name"), df("age") + 1).show()
// name    (age + 1)
// Michael null
// Andy    31
// Justin  20

// Select people older than 21
df.filter(df("age") > 21).show()
// age name
// 30  Andy

// Count people by age
df.groupBy("age").count().show()
// age  count
// null 1
// 19   1
// 30   1

val accidents = usagers
    .filter($"grav" == 2 ) // on filtre les victimes
    .groupBy("Num_Acc") // on groupe sur l'id de l'accident
    .count() // on compte le nombre des victimes
    .sort(desc("count"))// on trie par le nombre des victimes
    .limit(3)// on garde uniquement les 3 accidents les plus meurtirers
    .withColumnRenamed("count","nb_victimes")
    .join(caracteristiques, usagers("Num_acc")== caracteristiques("Num_Acc"))// on fait le jointure avec les caracteristiques
    .join(lieux, usagers("Num_acc")== lieux("Num_Acc"))// on fait le jointure avec les lieux
    .join(vehicules, usagers("Num_Acc")== vehicules("Num_Acc")) // on fait la jointure avec les vehicules
    .groupBy(caracteristiques("an"),caracteristiques("mois"),
            caracteristiques("jour"),caracteristiques("adr"),
            caracteristiques("dep"), caracteristiques("atm"),
            $"nb_victimes") // on groupe sur les colonnes qui nous interessent
    .count().withColumnRenamed("count","nb_vehicules") // on compte le nombre des vehicules

case class University(name: String, numStudents: Long, yearFounded: Long)

val schools = sqlContext.read.json("/schools.json").as[University]

schools.map(s => s"${s.name} is ${2015 – s.yearFounded} years old")

import spark.implicits._

// Our example case class Foo acting here as a schema
case class Foo(i: Long, j: String)

// Assuming spark is loaded and SparkSession is bind to spark
val initialDs = spark.createDataset( Foo(1, "Q") :: Foo(10, "W") :: Foo(100, "E") :: Nil )

// Assuming you are on Linux or Mac OS
initialDs.write.parquet("/tmp/foo")

val ds = spark.read.parquet("/tmp/foo").as[Foo]
// ds: org.apache.spark.sql.Dataset[Foo] = [i: bigint, j: string]

ds.show()
// +---+---+
// |  i|  j|
// +---+---+
// |  1|  Q|
// | 10|  W|
// |100|  E|
// +---+---+
//
//The value ds holds the content of the initialDs read from a parquet file. Let's try to only use field i from Foo and see how Spark's Catalyst (the query optimizer) optimizes this.

// Using a standard Spark TypedColumn in select()
val filteredDs = ds.filter($"i" === 10).select($"i".as[Long])
// filteredDs: org.apache.spark.sql.Dataset[Long] = [i: bigint]

filteredDs.show()
// +---+
// |  i|
// +---+
// | 10|
// +---+
//

// Using a standard Spark TypedColumn in select()
val filteredDs = ds.filter($"i" === 10).select($"i".as[Long])
// filteredDs: org.apache.spark.sql.Dataset[Long] = [i: bigint]

filteredDs.show()
// +---+
// |  i|
// +---+
// | 10|
// +---+
//

filteredDs.explain()
// == Physical Plan ==
// *Project [i#1771L]
// +- *Filter (isnotnull(i#1771L) && (i#1771L = 10))
//    +- *FileScan parquet [i#1771L] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/tmp/foo], PartitionFilters: [], PushedFilters: [IsNotNull(i), EqualTo(i,10)], ReadSchema: struct<i:bigint>

//Unfortunately, this syntax is not bulletproof: it fails at run-time if we try to access a non existing column x:

scala> ds.filter($"i" === 10).select($"x".as[Long])
org.apache.spark.sql.AnalysisException: cannot resolve '`x`' given input columns: [i, j];;
'Project ['x]
+- Filter (i#1771L = cast(10 as bigint))
   +- Relation[i#1771L,j#1772] parquet

//There are two things to improve:
// * we would want to avoid the as[Long] casting that we are required to type for type-safety.
// * we want a solution where reference to a non existing column name fails at compilation time

// Proposition

ds.filter(_.i == 10).map(_.i).show()
// +-----+
// |value|
// +-----+
// |   10|
// +-----+
//----

// The two closures in filter and map are functions that operate on Foo and the compiler will helps us capture all the mistakes we mentioned above.
   //
   //scala> ds.filter(_.i == 10).map(_.x).show()
   //<console>:20: error: value x is not a member of Foo
   //       ds.filter(_.i == 10).map(_.x).show()

// Unfortunately, this syntax does not allow Spark to optimize the code.

ds.filter(_.i == 10).map(_.i).explain()
// == Physical Plan ==
// *SerializeFromObject [input[0, bigint, false] AS value#1805L]
// +- *MapElements <function1>, obj#1804: bigint
//    +- *Filter <function1>.apply
//       +- *DeserializeToObject newInstance(class $line14.$read$$iw$$iw$$iw$$iw$Foo), obj#1803: $line14.$read$$iw$$iw$$iw$$iw$Foo
//          +- *FileScan parquet [i#1771L,j#1772] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/tmp/foo], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<i:bigint,j:string>

import frameless.TypedDataset
import frameless.syntax._

val fds = TypedDataset.create(ds)
// fds: frameless.TypedDataset[Foo] = [i: bigint, j: string]

fds.filter(fds('i) === 10).select(fds('i)).show().run()
// +---+
// | _1|
// +---+
// | 10|
// +---+
//

//And the optimized Physical Plan:

fds.filter(fds('i) === 10).select(fds('i)).explain()
// == Physical Plan ==
// *Project [i#1771L AS _1#1876L]
// +- *Filter (isnotnull(i#1771L) && (i#1771L = 10))
//    +- *FileScan parquet [i#1771L] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/tmp/foo], PartitionFilters: [], PushedFilters: [IsNotNull(i), EqualTo(i,10)], ReadSchema: struct<i:bigint>

//And the compiler is our friend.
scala> fds.filter(fds('i) === 10).select(fds('x))
<console>:24: error: No column Symbol with shapeless.tag.Tagged[String("x")] of type A in Foo
       fds.filter(fds('i) === 10).select(fds('x))
                                            ^