Spark 大数据分析平台


The Spark Big Data Analytics Platform Amir H. Payberah Swedish Institute of Computer Science amir@sics.se June 17, 2014 Amir H. Payberah (SICS) Spark June 17, 2014 1 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 2 / 125 I Big Data refers to datasets and flows large enough that has outpaced our capability to store, process, analyze, and understand. Amir H. Payberah (SICS) Spark June 17, 2014 3 / 125 Where Does Big Data Come From? Amir H. Payberah (SICS) Spark June 17, 2014 4 / 125 Big Data Market Driving Factors The number of web pages indexed by Google, which were around one million in 1998, have exceeded one trillion in 2008, and its expansion is accelerated by appearance of the social networks.∗ ∗“Mining big data: current status, and forecast to the future” [Wei Fan et al., 2013] Amir H. Payberah (SICS) Spark June 17, 2014 5 / 125 Big Data Market Driving Factors The amount of mobile data traffic is expected to grow to 10.8 Exabyte per month by 2016.∗ ∗“Worldwide Big Data Technology and Services 2012-2015 Forecast” [Dan Vesset et al., 2013] Amir H. Payberah (SICS) Spark June 17, 2014 6 / 125 Big Data Market Driving Factors More than 65 billion devices were connected to the Internet by 2010, and this number will go up to 230 billion by 2020.∗ ∗“The Internet of Things Is Coming” [John Mahoney et al., 2013] Amir H. Payberah (SICS) Spark June 17, 2014 7 / 125 Big Data Market Driving Factors Many companies are moving towards using Cloud services to access Big Data analytical tools. Amir H. Payberah (SICS) Spark June 17, 2014 8 / 125 Big Data Market Driving Factors Open source communities Amir H. Payberah (SICS) Spark June 17, 2014 9 / 125 How To Process Big Data? Amir H. Payberah (SICS) Spark June 17, 2014 10 / 125 Scale Up vs. Scale Out (1/2) I Scale up or scale vertically: adding resources to a single node in a system. I Scale out or scale horizontally: adding more nodes to a system. Amir H. Payberah (SICS) Spark June 17, 2014 11 / 125 Scale Up vs. Scale Out (2/2) I Scale up: more expensive than scaling out. I Scale out: more challenging for fault tolerance and software devel- opment. Amir H. Payberah (SICS) Spark June 17, 2014 12 / 125 Taxonomy of Parallel Architectures DeWitt, D. and Gray, J. “Parallel database systems: the future of high performance database systems”. ACM Communications, 35(6), 85-98, 1992. Amir H. Payberah (SICS) Spark June 17, 2014 13 / 125 Taxonomy of Parallel Architectures DeWitt, D. and Gray, J. “Parallel database systems: the future of high performance database systems”. ACM Communications, 35(6), 85-98, 1992. Amir H. Payberah (SICS) Spark June 17, 2014 13 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 14 / 125 Big Data Analytics Stack Amir H. Payberah (SICS) Spark June 17, 2014 15 / 125 Outline I Introduction to Scala I Data exploration using Spark I Stream processing with Spark Streaming I Graph analytics with GraphX Amir H. Payberah (SICS) Spark June 17, 2014 16 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 17 / 125 Scala I Scala: scalable language I A blend of object-oriented and functional programming I Runs on the Java Virtual Machine I Designed by Martin Odersky at EPFL Amir H. Payberah (SICS) Spark June 17, 2014 18 / 125 Functional Programming Languages I In a restricted sense: a language that does not have mutable vari- ables, assignments, or imperative control structures. I In a wider sense: it enables the construction of programs that focus on functions. I Functions are first-class citizens: • Defined anywhere (including inside other functions). • Passed as parameters to functions and returned as results. • Operators to compose functions. Amir H. Payberah (SICS) Spark June 17, 2014 19 / 125 Functional Programming Languages I In a restricted sense: a language that does not have mutable vari- ables, assignments, or imperative control structures. I In a wider sense: it enables the construction of programs that focus on functions. I Functions are first-class citizens: • Defined anywhere (including inside other functions). • Passed as parameters to functions and returned as results. • Operators to compose functions. Amir H. Payberah (SICS) Spark June 17, 2014 19 / 125 Scala Variables I Values: immutable I Variables: mutable var myVar: Int=0 val myVal: Int=1 I Scala data types: • Boolean, Byte, Short, Char, Int, Long, Float, Double, String Amir H. Payberah (SICS) Spark June 17, 2014 20 / 125 If ... Else varx= 30; if(x == 10){ println("Value of X is 10"); } else if(x == 20){ println("Value of X is 20"); } else{ println("This is else statement"); } Amir H. Payberah (SICS) Spark June 17, 2014 21 / 125 Loop vara=0 varb=0 for(a <-1 to3;b <-1 until3){ println("Value of a: "+a+ ", b: "+b) } // loop with collections val numList= List(1,2,3,4,5,6) for(a <- numList){ println("Value of a: "+a) } Amir H. Payberah (SICS) Spark June 17, 2014 22 / 125 Functions def functionName([list of parameters]):[return type]={ function body return[expr] } def addInt(a: Int, b: Int): Int={ var sum: Int=0 sum=a+b sum } println("Returned Value: "+ addInt(5,7)) Amir H. Payberah (SICS) Spark June 17, 2014 23 / 125 Anonymous Functions I Lightweight syntax for defining functions. var mul=(x: Int, y: Int) =>x*y println(mul(3,4)) Amir H. Payberah (SICS) Spark June 17, 2014 24 / 125 Higher-Order Functions def apply(f: Int => String, v: Int)=f(v) def layout(x: Int)= "["+x.toString()+ "]" println(apply(layout, 10)) Amir H. Payberah (SICS) Spark June 17, 2014 25 / 125 Collections (1/2) I Array: fixed-size sequential collection of elements of the same type valt= Array("zero", "one", "two") valb=t(0)// b = zero I List: sequential collection of elements of the same type valt= List("zero", "one", "two") valb=t(0)// b = zero I Set: sequential collection of elements of the same type without duplicates valt= Set("zero", "one", "two") valt.contains("zero") Amir H. Payberah (SICS) Spark June 17, 2014 26 / 125 Collections (1/2) I Array: fixed-size sequential collection of elements of the same type valt= Array("zero", "one", "two") valb=t(0)// b = zero I List: sequential collection of elements of the same type valt= List("zero", "one", "two") valb=t(0)// b = zero I Set: sequential collection of elements of the same type without duplicates valt= Set("zero", "one", "two") valt.contains("zero") Amir H. Payberah (SICS) Spark June 17, 2014 26 / 125 Collections (1/2) I Array: fixed-size sequential collection of elements of the same type valt= Array("zero", "one", "two") valb=t(0)// b = zero I List: sequential collection of elements of the same type valt= List("zero", "one", "two") valb=t(0)// b = zero I Set: sequential collection of elements of the same type without duplicates valt= Set("zero", "one", "two") valt.contains("zero") Amir H. Payberah (SICS) Spark June 17, 2014 26 / 125 Collections (2/2) I Map: collection of key/value pairs valm= Map(1 -> "sics",2 -> "kth") valb=m(1)// b = sics I Tuple:A fixed number of items of different types together valt=(1, "hello") valb=t._1 // b = 1 valc=t._2 // c = hello Amir H. Payberah (SICS) Spark June 17, 2014 27 / 125 Collections (2/2) I Map: collection of key/value pairs valm= Map(1 -> "sics",2 -> "kth") valb=m(1)// b = sics I Tuple:A fixed number of items of different types together valt=(1, "hello") valb=t._1 // b = 1 valc=t._2 // c = hello Amir H. Payberah (SICS) Spark June 17, 2014 27 / 125 Functional Combinators I map: applies a function over each element in the list val numbers= List(1,2,3,4) numbers.map(i =>i*2)// List(2, 4, 6, 8) I flatten: it collapses one level of nested structure List(List(1,2), List(3,4)).flatten // List(1, 2, 3, 4) I flatMap: map + flatten I foreach: it is like map but returns nothing Amir H. Payberah (SICS) Spark June 17, 2014 28 / 125 Classes and Objects class Calculator{ val brand: String= "HP" def add(m: Int,n: Int): Int=m+n } val calc= new Calculator calc.add(1,2) println(calc.brand) I A singleton is a class that can have only one instance. object Test{ def main(args: Array[String]){ ...} } Test.main(null) Amir H. Payberah (SICS) Spark June 17, 2014 29 / 125 Classes and Objects class Calculator{ val brand: String= "HP" def add(m: Int,n: Int): Int=m+n } val calc= new Calculator calc.add(1,2) println(calc.brand) I A singleton is a class that can have only one instance. object Test{ def main(args: Array[String]){ ...} } Test.main(null) Amir H. Payberah (SICS) Spark June 17, 2014 29 / 125 Case Classes and Pattern Matching I Case classes are used to store and match on the contents of a class. I They are designed to be used with pattern matching. I You can construct them without using new. case class Calc(brand: String, model: String) def calcType(calc: Calc)= calc match{ case Calc("hp", "20B") => "financial" case Calc("hp", "48G") => "scientific" case Calc("hp", "30B") => "business" case_ => "Calculator of unknown type" } calcType(Calc("hp", "20B")) Amir H. Payberah (SICS) Spark June 17, 2014 30 / 125 Simple Build Tool (SBT) I An open source build tool for Scala and Java projects. I Similar to Java’s Maven or Ant. I It is written in Scala. Amir H. Payberah (SICS) Spark June 17, 2014 31 / 125 SBT - Hello World! // make dir hello and edit Hello.scala object Hello{ def main(args: Array[String]){ println("Hello world.") } } $ cd hello $ sbt compile run Amir H. Payberah (SICS) Spark June 17, 2014 32 / 125 Common Commands I compile: compiles the main sources. I run *: run the main class. I package: creates a jar file. I console: starts the Scala interpreter. I clean: deletes all generated files. I help : displays detailed help for the specified command. Amir H. Payberah (SICS) Spark June 17, 2014 33 / 125 Create a Simple Project I Create project directory. I Create src/main/scala directory. I Create build.sbt in the project root. Amir H. Payberah (SICS) Spark June 17, 2014 34 / 125 build.sbt I A list of Scala expressions, separated by blank lines. I Located in the project’s base directory. $ cat build.sbt name := "hello" version := "1.0" scalaVersion := "2.10.4" Amir H. Payberah (SICS) Spark June 17, 2014 35 / 125 Add Dependencies I Add in build.sbt. I Module ID format: "groupID" %% "artifact" % "version" % "configuration" libraryDependencies += "org.apache.spark"%% "spark-core"% "1.0.0" // multiple dependencies libraryDependencies ++= Seq( "org.apache.spark"%% "spark-core"% "1.0.0", "org.apache.spark"%% "spark-streaming"% "1.0.0" ) I sbt uses the standard Maven2 repository by default, but you can add more resolvers. resolvers += "Akka Repository" at "http://repo.akka.io/releases/" Amir H. Payberah (SICS) Spark June 17, 2014 36 / 125 Scala Hands-on Exercises (1/4) I Declare a list of integers as a variable called myNumbers val myNumbers= List(1,2,5,4,7,3) I Declare a function, pow, that computes the second power of an Int def pow(a: Int): Int=a*a Amir H. Payberah (SICS) Spark June 17, 2014 37 / 125 Scala Hands-on Exercises (1/4) I Declare a list of integers as a variable called myNumbers val myNumbers= List(1,2,5,4,7,3) I Declare a function, pow, that computes the second power of an Int def pow(a: Int): Int=a*a Amir H. Payberah (SICS) Spark June 17, 2014 37 / 125 Scala Hands-on Exercises (1/4) I Declare a list of integers as a variable called myNumbers val myNumbers= List(1,2,5,4,7,3) I Declare a function, pow, that computes the second power of an Int def pow(a: Int): Int=a*a Amir H. Payberah (SICS) Spark June 17, 2014 37 / 125 Scala Hands-on Exercises (1/4) I Declare a list of integers as a variable called myNumbers val myNumbers= List(1,2,5,4,7,3) I Declare a function, pow, that computes the second power of an Int def pow(a: Int): Int=a*a Amir H. Payberah (SICS) Spark June 17, 2014 37 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (2/4) I Apply the function to myNumbers using the map function myNumbers.map(x => pow(x)) // or myNumbers.map(pow(_)) // or myNumbers.map(pow) I Write the pow function inline in a map call, using closure notation myNumbers.map(x =>x*x) I Iterate through myNumbers and print out its items for(i <- myNumbers) println(i) // or myNumbers.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 38 / 125 Scala Hands-on Exercises (3/4) I Declare a list of pair of string and integers as a variable called myList val myList= List[(String, Int)](("a",1),("b",2),("c",3)) I Write an inline function to increment the integer values of the list myList valx=v.map{ case(name, age) => age+1} // or valx=v.map(i =>i._2+1) // or valx=v.map(_._2+1) Amir H. Payberah (SICS) Spark June 17, 2014 39 / 125 Scala Hands-on Exercises (3/4) I Declare a list of pair of string and integers as a variable called myList val myList= List[(String, Int)](("a",1),("b",2),("c",3)) I Write an inline function to increment the integer values of the list myList valx=v.map{ case(name, age) => age+1} // or valx=v.map(i =>i._2+1) // or valx=v.map(_._2+1) Amir H. Payberah (SICS) Spark June 17, 2014 39 / 125 Scala Hands-on Exercises (3/4) I Declare a list of pair of string and integers as a variable called myList val myList= List[(String, Int)](("a",1),("b",2),("c",3)) I Write an inline function to increment the integer values of the list myList valx=v.map{ case(name, age) => age+1} // or valx=v.map(i =>i._2+1) // or valx=v.map(_._2+1) Amir H. Payberah (SICS) Spark June 17, 2014 39 / 125 Scala Hands-on Exercises (3/4) I Declare a list of pair of string and integers as a variable called myList val myList= List[(String, Int)](("a",1),("b",2),("c",3)) I Write an inline function to increment the integer values of the list myList valx=v.map{ case(name, age) => age+1} // or valx=v.map(i =>i._2+1) // or valx=v.map(_._2+1) Amir H. Payberah (SICS) Spark June 17, 2014 39 / 125 Scala Hands-on Exercises (4/4) I Do a word-count of a text file: create a Map with words as keys and counts of the number of occurrences of the word as values I You can load a text file as an array of lines as shown below: import scala.io.Source val lines= Source.fromFile("/root/spark/README.md").getLines.toArray I Then, instantiate a HashMap[String, Int] and use functional methods to populate it with word-counts val counts= new collection.mutable.HashMap[String, Int].withDefaultValue(0) lines.flatMap(_.split("""\W+""")).foreach(word => counts(word) +=1) counts.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 40 / 125 Scala Hands-on Exercises (4/4) I Do a word-count of a text file: create a Map with words as keys and counts of the number of occurrences of the word as values I You can load a text file as an array of lines as shown below: import scala.io.Source val lines= Source.fromFile("/root/spark/README.md").getLines.toArray I Then, instantiate a HashMap[String, Int] and use functional methods to populate it with word-counts val counts= new collection.mutable.HashMap[String, Int].withDefaultValue(0) lines.flatMap(_.split("""\W+""")).foreach(word => counts(word) +=1) counts.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 40 / 125 Scala Hands-on Exercises (4/4) I Do a word-count of a text file: create a Map with words as keys and counts of the number of occurrences of the word as values I You can load a text file as an array of lines as shown below: import scala.io.Source val lines= Source.fromFile("/root/spark/README.md").getLines.toArray I Then, instantiate a HashMap[String, Int] and use functional methods to populate it with word-counts val counts= new collection.mutable.HashMap[String, Int].withDefaultValue(0) lines.flatMap(_.split("""\W+""")).foreach(word => counts(word) +=1) counts.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 40 / 125 Scala Hands-on Exercises (4/4) I Do a word-count of a text file: create a Map with words as keys and counts of the number of occurrences of the word as values I You can load a text file as an array of lines as shown below: import scala.io.Source val lines= Source.fromFile("/root/spark/README.md").getLines.toArray I Then, instantiate a HashMap[String, Int] and use functional methods to populate it with word-counts val counts= new collection.mutable.HashMap[String, Int].withDefaultValue(0) lines.flatMap(_.split("""\W+""")).foreach(word => counts(word) +=1) counts.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 40 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 41 / 125 What is Spark? I An efficient distributed general-purpose data analysis platform. I Focusing on ease of programming and high performance. Amir H. Payberah (SICS) Spark June 17, 2014 42 / 125 Spark Big Data Analytics Stack Amir H. Payberah (SICS) Spark June 17, 2014 43 / 125 Motivation I MapReduce programming model has not been designed for complex operations, e.g., data mining. I Very expensive, i.e., always goes to disk and HDFS. Amir H. Payberah (SICS) Spark June 17, 2014 44 / 125 Solution I Extends MapReduce with more operators. I Support for advanced data flow graphs. I In-memory and out-of-core processing. Amir H. Payberah (SICS) Spark June 17, 2014 45 / 125 Spark vs. Hadoop Amir H. Payberah (SICS) Spark June 17, 2014 46 / 125 Spark vs. Hadoop Amir H. Payberah (SICS) Spark June 17, 2014 46 / 125 Spark vs. Hadoop Amir H. Payberah (SICS) Spark June 17, 2014 47 / 125 Spark vs. Hadoop Amir H. Payberah (SICS) Spark June 17, 2014 47 / 125 Resilient Distributed Datasets (RDD) (1/2) IA distributed memory abstraction. I Immutable collections of objects spread across a cluster. Amir H. Payberah (SICS) Spark June 17, 2014 48 / 125 Resilient Distributed Datasets (RDD) (1/2) IA distributed memory abstraction. I Immutable collections of objects spread across a cluster. Amir H. Payberah (SICS) Spark June 17, 2014 48 / 125 Resilient Distributed Datasets (RDD) (2/2) I An RDD is divided into a number of partitions, which are atomic pieces of information. I Partitions of an RDD can be stored on different nodes of a cluster. Amir H. Payberah (SICS) Spark June 17, 2014 49 / 125 RDD Operators I Higher-order functions: transformations and actions. I Transformations: lazy operators that create new RDDs. I Actions: launch a computation and return a value to the program or write data to the external storage. Amir H. Payberah (SICS) Spark June 17, 2014 50 / 125 Transformations vs. Actions Amir H. Payberah (SICS) Spark June 17, 2014 51 / 125 RDD Transformations - Map I All pairs are independently processed. // passing each element through a function. val nums= sc.parallelize(Array(1,2,3)) val squares= nums.map(x =>x*x)// {1, 4, 9} Amir H. Payberah (SICS) Spark June 17, 2014 52 / 125 RDD Transformations - Map I All pairs are independently processed. // passing each element through a function. val nums= sc.parallelize(Array(1,2,3)) val squares= nums.map(x =>x*x)// {1, 4, 9} Amir H. Payberah (SICS) Spark June 17, 2014 52 / 125 RDD Transformations - GroupBy I Pairs with identical key are grouped. I Groups are independently processed. val schools= sc.parallelize(Seq(("sics",1),("kth",1),("sics",2))) schools.groupByKey() // {("sics", (1, 2)), ("kth", (1))} schools.reduceByKey((x,y) =>x+y) // {("sics", 3), ("kth", 1)} Amir H. Payberah (SICS) Spark June 17, 2014 53 / 125 RDD Transformations - GroupBy I Pairs with identical key are grouped. I Groups are independently processed. val schools= sc.parallelize(Seq(("sics",1),("kth",1),("sics",2))) schools.groupByKey() // {("sics", (1, 2)), ("kth", (1))} schools.reduceByKey((x,y) =>x+y) // {("sics", 3), ("kth", 1)} Amir H. Payberah (SICS) Spark June 17, 2014 53 / 125 RDD Transformations - Join I Performs an equi-join on the key. I Join candidates are independently pro- cessed. val list1= sc.parallelize(Seq(("sics", "10"), ("kth", "50"), ("sics", "20"))) val list2= sc.parallelize(Seq(("sics", "upsala"), ("kth", "stockholm"))) list1.join(list2) // ("sics", ("10", "upsala")) // ("sics", ("20", "upsala")) // ("kth", ("50", "stockholm")) Amir H. Payberah (SICS) Spark June 17, 2014 54 / 125 RDD Transformations - Join I Performs an equi-join on the key. I Join candidates are independently pro- cessed. val list1= sc.parallelize(Seq(("sics", "10"), ("kth", "50"), ("sics", "20"))) val list2= sc.parallelize(Seq(("sics", "upsala"), ("kth", "stockholm"))) list1.join(list2) // ("sics", ("10", "upsala")) // ("sics", ("20", "upsala")) // ("kth", ("50", "stockholm")) Amir H. Payberah (SICS) Spark June 17, 2014 54 / 125 Basic RDD Actions I Return all the elements of the RDD as an array. val nums= sc.parallelize(Array(1,2,3)) nums.collect()// Array(1, 2, 3) I Return an array with the first n elements of the RDD. nums.take(2)// Array(1, 2) I Return the number of elements in the RDD. nums.count() // 3 I Aggregate the elements of the RDD using the given function. nums.reduce((x,y) =>x+y)// 6 Amir H. Payberah (SICS) Spark June 17, 2014 55 / 125 Basic RDD Actions I Return all the elements of the RDD as an array. val nums= sc.parallelize(Array(1,2,3)) nums.collect()// Array(1, 2, 3) I Return an array with the first n elements of the RDD. nums.take(2)// Array(1, 2) I Return the number of elements in the RDD. nums.count() // 3 I Aggregate the elements of the RDD using the given function. nums.reduce((x,y) =>x+y)// 6 Amir H. Payberah (SICS) Spark June 17, 2014 55 / 125 Basic RDD Actions I Return all the elements of the RDD as an array. val nums= sc.parallelize(Array(1,2,3)) nums.collect()// Array(1, 2, 3) I Return an array with the first n elements of the RDD. nums.take(2)// Array(1, 2) I Return the number of elements in the RDD. nums.count() // 3 I Aggregate the elements of the RDD using the given function. nums.reduce((x,y) =>x+y)// 6 Amir H. Payberah (SICS) Spark June 17, 2014 55 / 125 Basic RDD Actions I Return all the elements of the RDD as an array. val nums= sc.parallelize(Array(1,2,3)) nums.collect()// Array(1, 2, 3) I Return an array with the first n elements of the RDD. nums.take(2)// Array(1, 2) I Return the number of elements in the RDD. nums.count() // 3 I Aggregate the elements of the RDD using the given function. nums.reduce((x,y) =>x+y)// 6 Amir H. Payberah (SICS) Spark June 17, 2014 55 / 125 Creating RDDs I Turn a collection into an RDD. vala= sc.parallelize(Array(1,2,3)) I Load text file from local FS, HDFS, or S3. vala= sc.textFile("file.txt") valb= sc.textFile("directory/*.txt") valc= sc.textFile("hdfs://namenode:9000/path/file") Amir H. Payberah (SICS) Spark June 17, 2014 56 / 125 SparkContext I Main entry point to Spark functionality. I Available in shell as variable sc. I In standalone programs, you should make your own. import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ val sc= new SparkContext(master, appName,[sparkHome],[jars]) Amir H. Payberah (SICS) Spark June 17, 2014 57 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (1/3) I Read data from the given file hamlet and create an RDD named pagecounts val pagecounts= sc.textFile("hamlet") I Get the first 10 lines of hamlet pagecounts.take(10).foreach(println) I Count the total records in the data set pagecounts pagecounts.count Amir H. Payberah (SICS) Spark June 17, 2014 58 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (2/3) I Filter the data set pagecounts and return the items that have the word this, and cache in the memory val linesWithThis= pagecounts.filter(line => line.contains("this")).cache \\ or val linesWithThis= pagecounts.filter(_.contains("this")).cache I Find the lines with the most number of words. linesWithThis.map(line => line.split("").size) .reduce((a,b) => if(a>b)a elseb) I Count the total number of words val wordCounts= linesWithThis.flatMap(line => line.split("")).count \\ or val wordCounts= linesWithThis.flatMap(_.split("")).count Amir H. Payberah (SICS) Spark June 17, 2014 59 / 125 Spark Hands-on Exercises (3/3) I Count the number of distinct words val uniqueWordCounts= linesWithThis.flatMap(_.split("")).distinct.count I Count the number of each word val eachWordCounts= linesWithThis.flatMap(_.split("")) .map(word =>(word,1)) .reduceByKey((a,b) =>a+b) Amir H. Payberah (SICS) Spark June 17, 2014 60 / 125 Spark Hands-on Exercises (3/3) I Count the number of distinct words val uniqueWordCounts= linesWithThis.flatMap(_.split("")).distinct.count I Count the number of each word val eachWordCounts= linesWithThis.flatMap(_.split("")) .map(word =>(word,1)) .reduceByKey((a,b) =>a+b) Amir H. Payberah (SICS) Spark June 17, 2014 60 / 125 Spark Hands-on Exercises (3/3) I Count the number of distinct words val uniqueWordCounts= linesWithThis.flatMap(_.split("")).distinct.count I Count the number of each word val eachWordCounts= linesWithThis.flatMap(_.split("")) .map(word =>(word,1)) .reduceByKey((a,b) =>a+b) Amir H. Payberah (SICS) Spark June 17, 2014 60 / 125 Spark Hands-on Exercises (3/3) I Count the number of distinct words val uniqueWordCounts= linesWithThis.flatMap(_.split("")).distinct.count I Count the number of each word val eachWordCounts= linesWithThis.flatMap(_.split("")) .map(word =>(word,1)) .reduceByKey((a,b) =>a+b) Amir H. Payberah (SICS) Spark June 17, 2014 60 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 61 / 125 Motivation I Many applications must process large streams of live data and pro- vide results in real-time. I Processing information as it flows, without storing them persistently. I Traditional DBMSs: • Store and index data before processing it. • Process data only when explicitly asked by the users. • Both aspects contrast with our requirements. Amir H. Payberah (SICS) Spark June 17, 2014 62 / 125 Motivation I Many applications must process large streams of live data and pro- vide results in real-time. I Processing information as it flows, without storing them persistently. I Traditional DBMSs: • Store and index data before processing it. • Process data only when explicitly asked by the users. • Both aspects contrast with our requirements. Amir H. Payberah (SICS) Spark June 17, 2014 62 / 125 DBMS vs. DSMS (1/3) I DBMS: persistent data where updates are relatively infrequent. I DSMS: transient data that is continuously updated. Amir H. Payberah (SICS) Spark June 17, 2014 63 / 125 DBMS vs. DSMS (2/3) I DBMS: runs queries just once to return a complete answer. I DSMS: executes standing queries, which run continuously and pro- vide updated answers as new data arrives. Amir H. Payberah (SICS) Spark June 17, 2014 64 / 125 DBMS vs. DSMS (3/3) I Despite these differences, DSMSs resemble DBMSs: both process incoming data through a sequence of transformations based on SQL operators, e.g., selections, aggregates, joins. Amir H. Payberah (SICS) Spark June 17, 2014 65 / 125 Spark Streaming I Run a streaming computation as a series of very small, deterministic batch jobs. • Chop up the live stream into batches ofX seconds. • Spark treats each batch of data as RDDs and processes them using RDD operations. • Finally, the processed results of the RDD operations are returned in batches. Amir H. Payberah (SICS) Spark June 17, 2014 66 / 125 Spark Streaming I Run a streaming computation as a series of very small, deterministic batch jobs. • Chop up the live stream into batches ofX seconds. • Spark treats each batch of data as RDDs and processes them using RDD operations. • Finally, the processed results of the RDD operations are returned in batches. Amir H. Payberah (SICS) Spark June 17, 2014 66 / 125 Spark Streaming API (1/4) I DStream: sequence of RDDs representing a stream of data. • TCP sockets, Twitter, HDFS, Kafka, ... I Initializing Spark streaming val scc= new StreamingContext(master, appName, batchDuration, [sparkHome],[jars]) Amir H. Payberah (SICS) Spark June 17, 2014 67 / 125 Spark Streaming API (1/4) I DStream: sequence of RDDs representing a stream of data. • TCP sockets, Twitter, HDFS, Kafka, ... I Initializing Spark streaming val scc= new StreamingContext(master, appName, batchDuration, [sparkHome],[jars]) Amir H. Payberah (SICS) Spark June 17, 2014 67 / 125 Spark Streaming API (2/4) I Transformations: modify data from on DStream to a new DStream. • Standard RDD operations (stateless operations): map, join, ... • Stateful operations: group all the records from a sliding window of the past time intervals into one RDD: window, reduceByAndWindow, ... Window length: the duration of the window. Slide interval: the interval at which the operation is performed. Amir H. Payberah (SICS) Spark June 17, 2014 68 / 125 Spark Streaming API (2/4) I Transformations: modify data from on DStream to a new DStream. • Standard RDD operations (stateless operations): map, join, ... • Stateful operations: group all the records from a sliding window of the past time intervals into one RDD: window, reduceByAndWindow, ... Window length: the duration of the window. Slide interval: the interval at which the operation is performed. Amir H. Payberah (SICS) Spark June 17, 2014 68 / 125 Spark Streaming API (3/4) I Output operations: send data to external entity • saveAsHadoopFiles, foreach, print, ... I Attaching input sources ssc.textFileStream(directory) ssc.socketStream(hostname, port) Amir H. Payberah (SICS) Spark June 17, 2014 69 / 125 Spark Streaming API (3/4) I Output operations: send data to external entity • saveAsHadoopFiles, foreach, print, ... I Attaching input sources ssc.textFileStream(directory) ssc.socketStream(hostname, port) Amir H. Payberah (SICS) Spark June 17, 2014 69 / 125 Spark Streaming API (4/4) I Stream + Batch: It can be used to apply any RDD operation that is not exposed in the DStream API. val spamInfoRDD= sparkContext.hadoopFile(...) // join data stream with spam information to do data cleaning val cleanedDStream= inputDStream.transform(_.join(spamInfoRDD).filter(...)) I Stream + Interactive: Interactive queries on stream state from the Spark interpreter freqs.slice("21:00", "21:05").topK(10) I Starting/stopping the streaming computation ssc.start() ssc.stop() ssc.awaitTermination() Amir H. Payberah (SICS) Spark June 17, 2014 70 / 125 Spark Streaming API (4/4) I Stream + Batch: It can be used to apply any RDD operation that is not exposed in the DStream API. val spamInfoRDD= sparkContext.hadoopFile(...) // join data stream with spam information to do data cleaning val cleanedDStream= inputDStream.transform(_.join(spamInfoRDD).filter(...)) I Stream + Interactive: Interactive queries on stream state from the Spark interpreter freqs.slice("21:00", "21:05").topK(10) I Starting/stopping the streaming computation ssc.start() ssc.stop() ssc.awaitTermination() Amir H. Payberah (SICS) Spark June 17, 2014 70 / 125 Spark Streaming API (4/4) I Stream + Batch: It can be used to apply any RDD operation that is not exposed in the DStream API. val spamInfoRDD= sparkContext.hadoopFile(...) // join data stream with spam information to do data cleaning val cleanedDStream= inputDStream.transform(_.join(spamInfoRDD).filter(...)) I Stream + Interactive: Interactive queries on stream state from the Spark interpreter freqs.slice("21:00", "21:05").topK(10) I Starting/stopping the streaming computation ssc.start() ssc.stop() ssc.awaitTermination() Amir H. Payberah (SICS) Spark June 17, 2014 70 / 125 Example 1 (1/3) I Get hash-tags from Twitter. val ssc= new StreamingContext("local[2]", "Tweets", Seconds(1)) val tweets= TwitterUtils.createStream(ssc, None) DStream: a sequence of RDD representing a stream of data Amir H. Payberah (SICS) Spark June 17, 2014 71 / 125 Example 1 (2/3) I Get hash-tags from Twitter. val ssc= new StreamingContext("local[2]", "Tweets", Seconds(1)) val tweets= TwitterUtils.createStream(ssc, None) val hashTags= tweets.flatMap(status => getTags(status)) transformation: modify data in one DStream to create another DStream Amir H. Payberah (SICS) Spark June 17, 2014 72 / 125 Example 1 (3/3) I Get hash-tags from Twitter. val ssc= new StreamingContext("local[2]", "Tweets", Seconds(1)) val tweets= TwitterUtils.createStream(ssc, None) val hashTags= tweets.flatMap(status => getTags(status)) hashTags.saveAsHadoopFiles("hdfs://...") Amir H. Payberah (SICS) Spark June 17, 2014 73 / 125 Example 2 I Count frequency of words received every second. val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(1)) val lines= ssc.socketTextStream(ip, port) val words= lines.flatMap(_.split("")) val ones= words.map(x =>(x,1)) val freqs= ones.reduceByKey(_+_) Amir H. Payberah (SICS) Spark June 17, 2014 74 / 125 Example 3 I Count frequency of words received in last minute. val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(1)) val lines= ssc.socketTextStream(ip, port) val words= lines.flatMap(_.split("")) val ones= words.map(x =>(x,1)) val freqs= ones.reduceByKey(_+_) val freqs_60s= freqs.window(Seconds(60), Second(1)).reduceByKey(_+_) window length window movement Amir H. Payberah (SICS) Spark June 17, 2014 75 / 125 Example 3 - Simpler Model I Count frequency of words received in last minute. val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(1)) val lines= ssc.socketTextStream(ip, port) val words= lines.flatMap(_.split("")) val ones= words.map(x =>(x,1)) val freqs_60s= ones.reduceByKeyAndWindow(_+_, Seconds(60), Seconds(1)) Amir H. Payberah (SICS) Spark June 17, 2014 76 / 125 Example 3 - Incremental Window Operators I Count frequency of words received in last minute. // Associative only freqs_60s= ones.reduceByKeyAndWindow(_+_, Seconds(60), Seconds(1)) // Associative and invertible freqs_60s= ones.reduceByKeyAndWindow(_+_,_-_, Seconds(60), Seconds(1)) Associative only Associative and invertible Amir H. Payberah (SICS) Spark June 17, 2014 77 / 125 Spark Streaming Hands-on Exercises (1/2) I Stream data through a TCP connection and port 9999 nc -lk 9999 I import the streaming libraries import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ I Print out the incoming stream every five seconds at port 9999 val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) lines.print() Amir H. Payberah (SICS) Spark June 17, 2014 78 / 125 Spark Streaming Hands-on Exercises (1/2) I Stream data through a TCP connection and port 9999 nc -lk 9999 I import the streaming libraries import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ I Print out the incoming stream every five seconds at port 9999 val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) lines.print() Amir H. Payberah (SICS) Spark June 17, 2014 78 / 125 Spark Streaming Hands-on Exercises (1/2) I Stream data through a TCP connection and port 9999 nc -lk 9999 I import the streaming libraries import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ I Print out the incoming stream every five seconds at port 9999 val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) lines.print() Amir H. Payberah (SICS) Spark June 17, 2014 78 / 125 Spark Streaming Hands-on Exercises (1/2) I Stream data through a TCP connection and port 9999 nc -lk 9999 I import the streaming libraries import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ I Print out the incoming stream every five seconds at port 9999 val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) lines.print() Amir H. Payberah (SICS) Spark June 17, 2014 78 / 125 Spark Streaming Hands-on Exercises (1/2) I Stream data through a TCP connection and port 9999 nc -lk 9999 I import the streaming libraries import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ I Print out the incoming stream every five seconds at port 9999 val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) lines.print() Amir H. Payberah (SICS) Spark June 17, 2014 78 / 125 Spark Streaming Hands-on Exercises (1/2) I Count the number of each word in the incoming stream every five seconds at port 9999 import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(1)) val lines= ssc.socketTextStream("127.0.0.1", 9999) val words= lines.flatMap(_.split("")) val pairs= words.map(x =>(x,1)) val wordCounts= pairs.reduceByKey(_+_) wordCounts.print() Amir H. Payberah (SICS) Spark June 17, 2014 79 / 125 Spark Streaming Hands-on Exercises (1/2) I Count the number of each word in the incoming stream every five seconds at port 9999 import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(1)) val lines= ssc.socketTextStream("127.0.0.1", 9999) val words= lines.flatMap(_.split("")) val pairs= words.map(x =>(x,1)) val wordCounts= pairs.reduceByKey(_+_) wordCounts.print() Amir H. Payberah (SICS) Spark June 17, 2014 79 / 125 Spark Streaming Hands-on Exercises (2/2) I Extend the code to generate word count over last 30 seconds of data, and repeat the computation every 10 seconds import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) val words= lines.flatMap(_.split("")) val pairs= words.map(word =>(word,1)) val windowedWordCounts= pairs .reduceByKeyAndWindow(_+_,_-_, Seconds(30), Seconds(10)) windowedWordCounts.print() wordCounts.print() Amir H. Payberah (SICS) Spark June 17, 2014 80 / 125 Spark Streaming Hands-on Exercises (2/2) I Extend the code to generate word count over last 30 seconds of data, and repeat the computation every 10 seconds import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.StreamingContext._ val ssc= new StreamingContext("local[2]", "NetworkWordCount", Seconds(5)) val lines= ssc.socketTextStream("127.0.0.1", 9999) val words= lines.flatMap(_.split("")) val pairs= words.map(word =>(word,1)) val windowedWordCounts= pairs .reduceByKeyAndWindow(_+_,_-_, Seconds(30), Seconds(10)) windowedWordCounts.print() wordCounts.print() Amir H. Payberah (SICS) Spark June 17, 2014 80 / 125 Spark Streaming Hands-on Exercises - Twitter (1/7) I Twitter credential setup: to access Twitter’s sample tweet stream I Open the link: https://apps.twitter.com/ Amir H. Payberah (SICS) Spark June 17, 2014 81 / 125 Spark Streaming Hands-on Exercises - Twitter (1/7) I Twitter credential setup: to access Twitter’s sample tweet stream I Open the link: https://apps.twitter.com/ Amir H. Payberah (SICS) Spark June 17, 2014 81 / 125 Spark Streaming Hands-on Exercises - Twitter (2/7) Amir H. Payberah (SICS) Spark June 17, 2014 82 / 125 Spark Streaming Hands-on Exercises - Twitter (3/7) Amir H. Payberah (SICS) Spark June 17, 2014 83 / 125 Spark Streaming Hands-on Exercises - Twitter (4/7) Amir H. Payberah (SICS) Spark June 17, 2014 84 / 125 Spark Streaming Hands-on Exercises - Twitter (5/7) Amir H. Payberah (SICS) Spark June 17, 2014 85 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (6/7) I Create an StreamingContext for a batch duration of 5 seconds and use this context to create a stream of tweets import org.apache.spark.streaming.twitter._ val ssc= new StreamingContext("local[2]", "Tweets", Seconds(5)) val tweets= TwitterUtils.createStream(ssc, None) I Print the status text of the some of the tweets val statuses= tweets.map(status => status.getText()) statuses.print() I Get the stream of hashtags from the stream of tweets val words= statuses.flatMap(status => status.split("")) val hashtags= words.filter(word => word.startsWith("#")) hashtags.print() Amir H. Payberah (SICS) Spark June 17, 2014 86 / 125 Spark Streaming Hands-on Exercises - Twitter (7/7) I Set a path for periodic checkpointing of the intermediate data, and then count the hashtags over a one minute window ssc.checkpoint("/home/sics/temp") val counts= hashtags.map(tag =>(tag,1)) .reduceByKeyAndWindow(_+_,_-_, Seconds(60), Seconds(5)) counts.print() I Find the top 10 hashtags based on their counts val sortedCounts= counts.map{ case(tag, count) =>(count, tag)} .transform(rdd => rdd.sortByKey(false)) sortedCounts.foreachRDD(rdd => println("\nTop 10 hashtags:\n"+ rdd.take(10).mkString("\n"))) Amir H. Payberah (SICS) Spark June 17, 2014 87 / 125 Spark Streaming Hands-on Exercises - Twitter (7/7) I Set a path for periodic checkpointing of the intermediate data, and then count the hashtags over a one minute window ssc.checkpoint("/home/sics/temp") val counts= hashtags.map(tag =>(tag,1)) .reduceByKeyAndWindow(_+_,_-_, Seconds(60), Seconds(5)) counts.print() I Find the top 10 hashtags based on their counts val sortedCounts= counts.map{ case(tag, count) =>(count, tag)} .transform(rdd => rdd.sortByKey(false)) sortedCounts.foreachRDD(rdd => println("\nTop 10 hashtags:\n"+ rdd.take(10).mkString("\n"))) Amir H. Payberah (SICS) Spark June 17, 2014 87 / 125 Spark Streaming Hands-on Exercises - Twitter (7/7) I Set a path for periodic checkpointing of the intermediate data, and then count the hashtags over a one minute window ssc.checkpoint("/home/sics/temp") val counts= hashtags.map(tag =>(tag,1)) .reduceByKeyAndWindow(_+_,_-_, Seconds(60), Seconds(5)) counts.print() I Find the top 10 hashtags based on their counts val sortedCounts= counts.map{ case(tag, count) =>(count, tag)} .transform(rdd => rdd.sortByKey(false)) sortedCounts.foreachRDD(rdd => println("\nTop 10 hashtags:\n"+ rdd.take(10).mkString("\n"))) Amir H. Payberah (SICS) Spark June 17, 2014 87 / 125 Spark Streaming Hands-on Exercises - Twitter (7/7) I Set a path for periodic checkpointing of the intermediate data, and then count the hashtags over a one minute window ssc.checkpoint("/home/sics/temp") val counts= hashtags.map(tag =>(tag,1)) .reduceByKeyAndWindow(_+_,_-_, Seconds(60), Seconds(5)) counts.print() I Find the top 10 hashtags based on their counts val sortedCounts= counts.map{ case(tag, count) =>(count, tag)} .transform(rdd => rdd.sortByKey(false)) sortedCounts.foreachRDD(rdd => println("\nTop 10 hashtags:\n"+ rdd.take(10).mkString("\n"))) Amir H. Payberah (SICS) Spark June 17, 2014 87 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 88 / 125 Amir H. Payberah (SICS) Spark June 17, 2014 89 / 125 Introduction I Graphs provide a flexible abstraction for describing relationships be- tween discrete objects. I Many problems can be modeled by graphs and solved with appro- priate graph algorithms. Amir H. Payberah (SICS) Spark June 17, 2014 90 / 125 Large Graph Amir H. Payberah (SICS) Spark June 17, 2014 91 / 125 Large-Scale Graph Processing I Large graphs need large-scale processing. I A large graph either cannot fit into memory of single computer or it fits with huge cost. Amir H. Payberah (SICS) Spark June 17, 2014 92 / 125 Question Can we use platforms like MapReduce or Spark, which are based on data-parallel model, for large-scale graph proceeding? Amir H. Payberah (SICS) Spark June 17, 2014 93 / 125 Data-Parallel Model for Large-Scale Graph Processing I The platforms that have worked well for developing parallel applica- tions are not necessarily effective for large-scale graph problems. I Why? Amir H. Payberah (SICS) Spark June 17, 2014 94 / 125 Graph Algorithms Characteristics (1/2) I Unstructured problems • Difficult to extract parallelism based on partitioning of the data: the irregular structure of graphs. • Limited scalability: unbalanced computational loads resulting from poorly partitioned data. I Data-driven computations • Difficult to express parallelism based on partitioning of computation: the structure of computations in the algorithm is not known a priori. • The computations are dictated by nodes and links of the graph. Amir H. Payberah (SICS) Spark June 17, 2014 95 / 125 Graph Algorithms Characteristics (1/2) I Unstructured problems • Difficult to extract parallelism based on partitioning of the data: the irregular structure of graphs. • Limited scalability: unbalanced computational loads resulting from poorly partitioned data. I Data-driven computations • Difficult to express parallelism based on partitioning of computation: the structure of computations in the algorithm is not known a priori. • The computations are dictated by nodes and links of the graph. Amir H. Payberah (SICS) Spark June 17, 2014 95 / 125 Graph Algorithms Characteristics (2/2) I Poor data locality • The computations and data access patterns do not have much local- ity: the irregular structure of graphs. I High data access to computation ratio • Graph algorithms are often based on exploring the structure of a graph to perform computations on the graph data. • Runtime can be dominated by waiting memory fetches: low locality. Amir H. Payberah (SICS) Spark June 17, 2014 96 / 125 Graph Algorithms Characteristics (2/2) I Poor data locality • The computations and data access patterns do not have much local- ity: the irregular structure of graphs. I High data access to computation ratio • Graph algorithms are often based on exploring the structure of a graph to perform computations on the graph data. • Runtime can be dominated by waiting memory fetches: low locality. Amir H. Payberah (SICS) Spark June 17, 2014 96 / 125 Proposed Solution Graph-Parallel Processing I Computation typically depends on the neighbors. Amir H. Payberah (SICS) Spark June 17, 2014 97 / 125 Proposed Solution Graph-Parallel Processing I Computation typically depends on the neighbors. Amir H. Payberah (SICS) Spark June 17, 2014 97 / 125 Graph-Parallel Processing I Restricts the types of computation. I New techniques to partition and distribute graphs. I Exploit graph structure. I Executes graph algorithms orders-of-magnitude faster than more general data-parallel systems. Amir H. Payberah (SICS) Spark June 17, 2014 98 / 125 Data-Parallel vs. Graph-Parallel Computation Amir H. Payberah (SICS) Spark June 17, 2014 99 / 125 Data-Parallel vs. Graph-Parallel Computation I Data-parallel computation • Record-centric view of data. • Parallelism: processing independent data on separate resources. I Graph-parallel computation • Vertex-centric view of graphs. • Parallelism: partitioning graph (dependent) data across processing resources, and resolving dependencies(along edges) through iterative computation and communication. Amir H. Payberah (SICS) Spark June 17, 2014 100 / 125 Graph-Parallel Computation Frameworks Amir H. Payberah (SICS) Spark June 17, 2014 101 / 125 Data-Parallel vs. Graph-Parallel Computation I Graph-parallel computation: restricting the types of computation to achieve performance. I But, the same restrictions make it difficult and inefficient to express many stages in a typical graph-analytics pipeline. Amir H. Payberah (SICS) Spark June 17, 2014 102 / 125 Data-Parallel vs. Graph-Parallel Computation I Graph-parallel computation: restricting the types of computation to achieve performance. I But, the same restrictions make it difficult and inefficient to express many stages in a typical graph-analytics pipeline. Amir H. Payberah (SICS) Spark June 17, 2014 102 / 125 Data-Parallel and Graph-Parallel Pipeline I Moving between table and graph views of the same physical data. I Inefficient: extensive data movement and duplication across the net- work and file system. Amir H. Payberah (SICS) Spark June 17, 2014 103 / 125 GraphX vs. Data-Parallel/Graph-Parallel Systems Amir H. Payberah (SICS) Spark June 17, 2014 104 / 125 GraphX vs. Data-Parallel/Graph-Parallel Systems Amir H. Payberah (SICS) Spark June 17, 2014 104 / 125 GraphX I New API that blurs the distinction between Tables and Graphs. I New system that unifies Data-Parallel and Graph-Parallel systems. I It is implemented on top of Spark. Amir H. Payberah (SICS) Spark June 17, 2014 105 / 125 Unifying Data-Parallel and Graph-Parallel Analytics I Tables and Graphs are composable views of the same physical data. I Each view has its own operators that exploit the semantics of the view to achieve efficient execution. Amir H. Payberah (SICS) Spark June 17, 2014 106 / 125 Data Model I Property Graph: represented using two Spark RDDs: • Edge collection: VertexRDD • Vertex collection: EdgeRDD // VD: the type of the vertex attribute // ED: the type of the edge attribute class Graph[VD,ED]{ val vertices: VertexRDD[VD] val edges: EdgeRDD[ED,VD] } Amir H. Payberah (SICS) Spark June 17, 2014 107 / 125 Primitive Data Types // Vertex collection class VertexRDD[VD] extends RDD[(VertexId,VD)] // Edge collection class EdgeRDD[ED] extends RDD[Edge[ED]] case class Edge[ED,VD](srcId: VertexId=0, dstId: VertexId=0, attr:ED= null.asInstanceOf[ED]) // Edge Triple class EdgeTriplet[VD,ED] extends Edge[ED] I EdgeTriplet represents an edge along with the vertex attributes of its neighboring vertices. Amir H. Payberah (SICS) Spark June 17, 2014 108 / 125 Example (1/3) Amir H. Payberah (SICS) Spark June 17, 2014 109 / 125 Example (2/3) val sc: SparkContext // Create an RDD for the vertices val users: RDD[(Long,(String, String))]= sc.parallelize( Array((3L,("rxin", "student")),(7L,("jgonzal", "postdoc")), (5L,("franklin", "prof")),(2L,("istoica", "prof")))) // Create an RDD for edges val relationships: RDD[Edge[String]]= sc.parallelize( Array(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"), Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi"))) // Define a default user in case there are relationship with missing user val defaultUser=("John Doe", "Missing") // Build the initial Graph val userGraph: Graph[(String, String), String]= Graph(users, relationships, defaultUser) Amir H. Payberah (SICS) Spark June 17, 2014 110 / 125 Example (3/3) // Constructed from above val userGraph: Graph[(String, String), String] // Count all users which are postdocs userGraph.vertices.filter{ case(id,(name, pos)) => pos == "postdoc"}.count // Count all the edges where src > dst userGraph.edges.filter(e =>e.srcId>e.dstId).count // Use the triplets view to create an RDD of facts val facts: RDD[String]= graph.triplets.map(triplet => triplet.srcAttr._1+ " is the "+ triplet.attr+ " of "+ triplet.dstAttr._1) // Remove missing vertices as well as the edges to connected to them val validGraph= graph.subgraph(vpred=(id, attr) => attr._2 != "Missing") facts.collect.foreach(println) Amir H. Payberah (SICS) Spark June 17, 2014 111 / 125 Property Operators (1/2) class Graph[VD,ED]{ def mapVertices[VD2](map:(VertexId,VD) => VD2): Graph[VD2, ED] def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2] def mapTriplets[ED2](map: EdgeTriplet[VD,ED] => ED2): Graph[VD, ED2] } I They yield new graphs with the vertex or edge properties modified by the map function. I The graph structure is unaffected. Amir H. Payberah (SICS) Spark June 17, 2014 112 / 125 Property Operators (2/2) val newGraph= graph.mapVertices((id, attr) => mapUdf(id, attr)) val newVertices= graph.vertices.map((id, attr) =>(id, mapUdf(id, attr))) val newGraph= Graph(newVertices, graph.edges) I Both are logically equivalent, but the second one does not preserve the structural indices and would not benefit from the GraphX system optimizations. Amir H. Payberah (SICS) Spark June 17, 2014 113 / 125 Map Reduce Triplets I Map-Reduce for each vertex // what is the age of the oldest follower for each user? val oldestFollowerAge= graph.mapReduceTriplets( e => Iterator((e.dstAttr,e.srcAttr)), // Map (a,b) => max(a,b)// Reduce ).vertices Amir H. Payberah (SICS) Spark June 17, 2014 114 / 125 Map Reduce Triplets I Map-Reduce for each vertex // what is the age of the oldest follower for each user? val oldestFollowerAge= graph.mapReduceTriplets( e => Iterator((e.dstAttr,e.srcAttr)), // Map (a,b) => max(a,b)// Reduce ).vertices Amir H. Payberah (SICS) Spark June 17, 2014 114 / 125 Structural Operators class Graph[VD,ED]{ // returns a new graph with all the edge directions reversed def reverse: Graph[VD,ED] // returns the graph containing only the vertices and edges that satisfy // the vertex predicate def subgraph(epred: EdgeTriplet[VD,ED] => Boolean, vpred:(VertexId,VD) => Boolean): Graph[VD,ED] // a subgraph by returning a graph that contains the vertices and edges // that are also found in the input graph def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED] } Amir H. Payberah (SICS) Spark June 17, 2014 115 / 125 Structural Operators Example // Build the initial Graph val graph= Graph(users, relationships, defaultUser) // Run Connected Components val ccGraph= graph.connectedComponents() // Remove missing vertices as well as the edges to connected to them val validGraph= graph.subgraph(vpred=(id, attr) => attr._2 != "Missing") // Restrict the answer to the valid subgraph val validCCGraph= ccGraph.mask(validGraph) Amir H. Payberah (SICS) Spark June 17, 2014 116 / 125 Join Operators I To join data from external collections (RDDs) with graphs. class Graph[VD,ED]{ // joins the vertices with the input RDD and returns a new graph // by applying the map function to the result of the joined vertices def joinVertices[U](table: RDD[(VertexId,U)]) (map:(VertexId,VD,U) =>VD): Graph[VD,ED] // similarly to joinVertices, but the map function is applied to // all vertices and can change the vertex property type def outerJoinVertices[U, VD2](table: RDD[(VertexId,U)]) (map:(VertexId,VD, Option[U]) => VD2): Graph[VD2,ED] } Amir H. Payberah (SICS) Spark June 17, 2014 117 / 125 GraphX Hands-on Exercises (1/7) Amir H. Payberah (SICS) Spark June 17, 2014 118 / 125 GraphX Hands-on Exercises (2/7) I import the streaming libraries import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD I Build the property graph shown in the last page val vertexArray= Array( (1L,("Alice", 28)),(2L,("Bob", 27)),(3L,("Charlie", 65)), (4L,("David", 42)),(5L,("Ed", 55)),(6L,("Fran", 50))) val edgeArray= Array( Edge(2L, 1L,7), Edge(2L, 4L,2), Edge(3L, 2L,4), Edge(3L, 6L,3), Edge(4L, 1L,1), Edge(5L, 2L,2), Edge(5L, 3L,8), Edge(5L, 6L,3)) val vertexRDD: RDD[(Long,(String, Int))]= sc.parallelize(vertexArray) val edgeRDD: RDD[Edge[Int]]= sc.parallelize(edgeArray) val graph: Graph[(String, Int), Int]= Graph(vertexRDD, edgeRDD) Amir H. Payberah (SICS) Spark June 17, 2014 119 / 125 GraphX Hands-on Exercises (2/7) I import the streaming libraries import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD I Build the property graph shown in the last page val vertexArray= Array( (1L,("Alice", 28)),(2L,("Bob", 27)),(3L,("Charlie", 65)), (4L,("David", 42)),(5L,("Ed", 55)),(6L,("Fran", 50))) val edgeArray= Array( Edge(2L, 1L,7), Edge(2L, 4L,2), Edge(3L, 2L,4), Edge(3L, 6L,3), Edge(4L, 1L,1), Edge(5L, 2L,2), Edge(5L, 3L,8), Edge(5L, 6L,3)) val vertexRDD: RDD[(Long,(String, Int))]= sc.parallelize(vertexArray) val edgeRDD: RDD[Edge[Int]]= sc.parallelize(edgeArray) val graph: Graph[(String, Int), Int]= Graph(vertexRDD, edgeRDD) Amir H. Payberah (SICS) Spark June 17, 2014 119 / 125 GraphX Hands-on Exercises (2/7) I import the streaming libraries import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD I Build the property graph shown in the last page val vertexArray= Array( (1L,("Alice", 28)),(2L,("Bob", 27)),(3L,("Charlie", 65)), (4L,("David", 42)),(5L,("Ed", 55)),(6L,("Fran", 50))) val edgeArray= Array( Edge(2L, 1L,7), Edge(2L, 4L,2), Edge(3L, 2L,4), Edge(3L, 6L,3), Edge(4L, 1L,1), Edge(5L, 2L,2), Edge(5L, 3L,8), Edge(5L, 6L,3)) val vertexRDD: RDD[(Long,(String, Int))]= sc.parallelize(vertexArray) val edgeRDD: RDD[Edge[Int]]= sc.parallelize(edgeArray) val graph: Graph[(String, Int), Int]= Graph(vertexRDD, edgeRDD) Amir H. Payberah (SICS) Spark June 17, 2014 119 / 125 GraphX Hands-on Exercises (2/7) I import the streaming libraries import org.apache.spark.graphx._ import org.apache.spark.rdd.RDD I Build the property graph shown in the last page val vertexArray= Array( (1L,("Alice", 28)),(2L,("Bob", 27)),(3L,("Charlie", 65)), (4L,("David", 42)),(5L,("Ed", 55)),(6L,("Fran", 50))) val edgeArray= Array( Edge(2L, 1L,7), Edge(2L, 4L,2), Edge(3L, 2L,4), Edge(3L, 6L,3), Edge(4L, 1L,1), Edge(5L, 2L,2), Edge(5L, 3L,8), Edge(5L, 6L,3)) val vertexRDD: RDD[(Long,(String, Int))]= sc.parallelize(vertexArray) val edgeRDD: RDD[Edge[Int]]= sc.parallelize(edgeArray) val graph: Graph[(String, Int), Int]= Graph(vertexRDD, edgeRDD) Amir H. Payberah (SICS) Spark June 17, 2014 119 / 125 GraphX Hands-on Exercises (3/7) I Display the name of the users older than 30 years old graph.vertices.filter{ case(id,(name, age)) => age> 30}.foreach{ case(id,(name, age)) => println(s"$name is $age") } I Display who follows who (through the edges direction). /** * Triplet has the following Fields: * triplet.srcAttr: (String, Int) * triplet.dstAttr: (String, Int) * triplet.attr: Int * triplet.srcId: VertexId * triplet.dstId: VertexId */ graph.triplets.foreach(t => println(s"${t.srcAttr._1} follows ${t.dstAttr._1}")) Amir H. Payberah (SICS) Spark June 17, 2014 120 / 125 GraphX Hands-on Exercises (3/7) I Display the name of the users older than 30 years old graph.vertices.filter{ case(id,(name, age)) => age> 30}.foreach{ case(id,(name, age)) => println(s"$name is $age") } I Display who follows who (through the edges direction). /** * Triplet has the following Fields: * triplet.srcAttr: (String, Int) * triplet.dstAttr: (String, Int) * triplet.attr: Int * triplet.srcId: VertexId * triplet.dstId: VertexId */ graph.triplets.foreach(t => println(s"${t.srcAttr._1} follows ${t.dstAttr._1}")) Amir H. Payberah (SICS) Spark June 17, 2014 120 / 125 GraphX Hands-on Exercises (3/7) I Display the name of the users older than 30 years old graph.vertices.filter{ case(id,(name, age)) => age> 30}.foreach{ case(id,(name, age)) => println(s"$name is $age") } I Display who follows who (through the edges direction). /** * Triplet has the following Fields: * triplet.srcAttr: (String, Int) * triplet.dstAttr: (String, Int) * triplet.attr: Int * triplet.srcId: VertexId * triplet.dstId: VertexId */ graph.triplets.foreach(t => println(s"${t.srcAttr._1} follows ${t.dstAttr._1}")) Amir H. Payberah (SICS) Spark June 17, 2014 120 / 125 GraphX Hands-on Exercises (3/7) I Display the name of the users older than 30 years old graph.vertices.filter{ case(id,(name, age)) => age> 30}.foreach{ case(id,(name, age)) => println(s"$name is $age") } I Display who follows who (through the edges direction). /** * Triplet has the following Fields: * triplet.srcAttr: (String, Int) * triplet.dstAttr: (String, Int) * triplet.attr: Int * triplet.srcId: VertexId * triplet.dstId: VertexId */ graph.triplets.foreach(t => println(s"${t.srcAttr._1} follows ${t.dstAttr._1}")) Amir H. Payberah (SICS) Spark June 17, 2014 120 / 125 GraphX Hands-on Exercises (4/7) I Compute the total age of followers of each user and print them out val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator(...), // map (a,b) => ... // reduce ) val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator((triplet.dstId, triplet.srcAttr._2)), (a,b) =>a+b) followers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 121 / 125 GraphX Hands-on Exercises (4/7) I Compute the total age of followers of each user and print them out val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator(...), // map (a,b) => ... // reduce ) val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator((triplet.dstId, triplet.srcAttr._2)), (a,b) =>a+b) followers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 121 / 125 GraphX Hands-on Exercises (4/7) I Compute the total age of followers of each user and print them out val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator(...), // map (a,b) => ... // reduce ) val followers: VertexRDD[Int]= graph.mapReduceTriplets[Int]( triplet => Iterator((triplet.dstId, triplet.srcAttr._2)), (a,b) =>a+b) followers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 121 / 125 GraphX Hands-on Exercises (5/7) I Compute the average age of followers of each user and print them out val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator(...), // map (a,b) => (...) // reduce ) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues(...) val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator((triplet.dstId,(1, triplet.srcAttr._2))), (a,b) =>(a._1+b._1,a._2+b._2)) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues((id, value) => value match{ case(count, totalAge) => totalAge/ count }) avgAgeOfFollowers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 122 / 125 GraphX Hands-on Exercises (5/7) I Compute the average age of followers of each user and print them out val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator(...), // map (a,b) => (...) // reduce ) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues(...) val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator((triplet.dstId,(1, triplet.srcAttr._2))), (a,b) =>(a._1+b._1,a._2+b._2)) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues((id, value) => value match{ case(count, totalAge) => totalAge/ count }) avgAgeOfFollowers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 122 / 125 GraphX Hands-on Exercises (5/7) I Compute the average age of followers of each user and print them out val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator(...), // map (a,b) => (...) // reduce ) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues(...) val followers: VertexRDD[(Int, Double)]= graph .mapReduceTriplets[(Int, Double)]( triplet => Iterator((triplet.dstId,(1, triplet.srcAttr._2))), (a,b) =>(a._1+b._1,a._2+b._2)) val avgAgeOfFollowers: VertexRDD[Double]= followers.mapValues((id, value) => value match{ case(count, totalAge) => totalAge/ count }) avgAgeOfFollowers.collect.foreach(print) Amir H. Payberah (SICS) Spark June 17, 2014 122 / 125 GraphX Hands-on Exercises (6/7) I Make a subgraph of the users that are 30 or older val olderGraph= graph.subgraph(vpred= ...) val olderGraph= graph.subgraph(vpred=(id,u) =>u._2 >= 30) Amir H. Payberah (SICS) Spark June 17, 2014 123 / 125 GraphX Hands-on Exercises (6/7) I Make a subgraph of the users that are 30 or older val olderGraph= graph.subgraph(vpred= ...) val olderGraph= graph.subgraph(vpred=(id,u) =>u._2 >= 30) Amir H. Payberah (SICS) Spark June 17, 2014 123 / 125 GraphX Hands-on Exercises (6/7) I Make a subgraph of the users that are 30 or older val olderGraph= graph.subgraph(vpred= ...) val olderGraph= graph.subgraph(vpred=(id,u) =>u._2 >= 30) Amir H. Payberah (SICS) Spark June 17, 2014 123 / 125 GraphX Hands-on Exercises (7/7) I Compute the connected components and display the component id of each user in oldGraph val cc= olderGraph... olderGraph.vertices.leftJoin(cc.vertices){ ... }.foreach{...} val cc= olderGraph.connectedComponents olderGraph.vertices.leftJoin(cc.vertices){ case(id,u, comp) =>s"${u._1} is in component ${comp.get}" }.foreach{ case(id, str) => println(str)} Amir H. Payberah (SICS) Spark June 17, 2014 124 / 125 GraphX Hands-on Exercises (7/7) I Compute the connected components and display the component id of each user in oldGraph val cc= olderGraph... olderGraph.vertices.leftJoin(cc.vertices){ ... }.foreach{...} val cc= olderGraph.connectedComponents olderGraph.vertices.leftJoin(cc.vertices){ case(id,u, comp) =>s"${u._1} is in component ${comp.get}" }.foreach{ case(id, str) => println(str)} Amir H. Payberah (SICS) Spark June 17, 2014 124 / 125 GraphX Hands-on Exercises (7/7) I Compute the connected components and display the component id of each user in oldGraph val cc= olderGraph... olderGraph.vertices.leftJoin(cc.vertices){ ... }.foreach{...} val cc= olderGraph.connectedComponents olderGraph.vertices.leftJoin(cc.vertices){ case(id,u, comp) =>s"${u._1} is in component ${comp.get}" }.foreach{ case(id, str) => println(str)} Amir H. Payberah (SICS) Spark June 17, 2014 124 / 125 Questions? Amir H. Payberah (SICS) Spark June 17, 2014 125 / 125

下载pdf到电脑,查找使用更方便

pdf的实际排版效果,会与网站的显示效果略有不同!!

需要 8 金币 [ 分享pdf获得金币 ] 3 人已下载

下载pdf

pdf贡献者

bw3e

贡献于2015-03-11

下载需要 8 金币 [金币充值 ]
亲,您也可以通过 分享原创pdf 来获得金币奖励!
下载pdf