Apache Spark RDD API Examples

Apache Spark RDD API Examples
an RDD is actually more than that. On cluster installations, separate data partitions can be on separate nodes. Using the RDD as a handle one can access all partitions and perform computations and transformations using the contained data. Whenever a part of a RDD or an entire RDD is lost, the system is able to reconstruct the data of lost partitions by using lineage information. Lineage refers to the sequence of transformations used to produce the current RDD. As a result, Spark is able to recover automatically from most failures.

All RDDs available in Spark derive either directly or indirectly from the class RDD. This class comes with a large set of methods that perform operations on the data within the associated partitions. The class RDD is abstract. Whenever, one uses a RDD, one is actually using a concertized implementation of RDD. These implementations have to overwrite some core functions to make the RDD behave as expected.

It does not impose restrictions regarding what data can be stored within RDD partitions.

The basic RDD API considers each data item as a single value. However, users often want to work with key-value pairs. Therefore Spark extended the interface of RDD to provide additional functions (PairRDDFunctions), which explicitly work on key-value pairs. Currently, there are four extensions to the RDD API available in spark. They are as follows:

DoubleRDDFunctions 

This extension contains many useful methods for aggregating numeric values. They become available if the data items of an RDD are implicitly convertible to the Scala data-type double.

PairRDDFunctions 

Methods defined in this interface extension become available when the data items have a two component tuple structure. Spark will interpret the first tuple item (i.e. tuplename. 1) as the key and the second item (i.e. tuplename. 2) as the associated value.

OrderedRDDFunctions 

Methods defined in this interface extension become available if the data items are two-component tuples where the key is implicitly sortable.

SequenceFileRDDFunctions 

This extension contains several methods that allow users to create Hadoop sequence- les from RDDs. The data items must be two compo- nentkey-value tuples as required by the PairRDDFunctions. However, there are additional requirements considering the convertibility of the tuple components to Writable types.

val z = sc.parallelize(List(1,2,3,4,5,6), 2)
z.aggregate(0)(math.max(_, _), _ + _)

sortBy

This function sorts the input RDD's data and stores it in a new RDD. The first parameter requires you to specify a function which  maps the input data into the key that you want to sortBy. The second parameter (optional) specifies whether you want the data to be sorted in ascending or descending order.
val y = sc.parallelize(Array(5, 7, 1, 3, 2, 1))
y.sortBy(c => c, true).collect
val z = sc.parallelize(Array(("H", 10), ("A", 26), ("Z", 1), ("L", 5)))
z.sortBy(c => c._1, true).collect
z.sortBy(c => c._2, true).collect

val x = sc.parallelize(List(1.0, 2.0, 3.0, 5.0, 20.0, 19.02, 19.29, 11.09, 21.0), 2)
x.stats

sortByKey [Ordered]

This function sorts the input RDD's data and stores it in a new RDD. The output RDD is a shuffled RDD because it stores data that is output by a reducer which has been shuffled. The implementation of this function is actually very clever. First, it uses a range partitioner to partition the data in ranges within the shuffled RDD. Then it sorts these ranges individually with mapPartitions using standard sort mechanisms.
val a = sc.parallelize(List("dog", "cat", "owl", "gnu", "ant"), 2)
val b = sc.parallelize(1 to a.count.toInt, 2)
val c = a.zip(b)
c.sortByKey(true).collect
res74: Array[(String, Int)] = Array((ant,5), (cat,2), (dog,1), (gnu,4), (owl,3))
c.sortByKey(false).collect
res75: Array[(String, Int)] = Array((owl,3), (gnu,4), (dog,1), (cat,2), (ant,5))

val a = sc.parallelize(1 to 100, 5)
val b = a.cartesian(a)
val c = sc.parallelize(b.takeSample(true, 5, 13), 2)
val d = c.sortByKey(false)
res56: Array[(Int, Int)] = Array((96,9), (84,76), (59,59), (53,65), (52,4))

val x = sc.parallelize(List(1.0, 2.0, 3.0, 5.0, 20.0, 19.02, 19.29, 11.09, 21.0), 2)
x.sum
takeOrdered

Orders the data items of the RDD using their inherent implicit ordering function and returns the first n items as an array.

val b = sc.parallelize(List("dog", "cat", "ape", "salmon", "gnu"), 2) b.takeOrdered(2)

takeSample

Behaves different from sample in the following respects:

•   It will return an exact number of samples (Hint: 2nd parameter)
•   It returns an Array instead of RDD.
•   It internally randomizes the order of the items returned.

val x = sc.parallelize(1 to 1000, 3)
x.takeSample(true, 100, 1)
toDebugString

val a = sc.parallelize(1 to 9, 3) val b = sc.parallelize(1 to 3, 3) val c = a.subtract(b) c.toDebugString

top

Utilizes the implicit ordering of $T$ to determine the top $k$ values and returns them as an array.
union, ++

Performs the standard set operation: A union B

val a = sc.parallelize(1 to 3, 1) val b = sc.parallelize(5 to 7, 1) (a ++ b).collect

unpersist

Dematerializes the RDD (i.e. Erases all data items from hard-disk and memory). However, the RDD object remains. If it is referenced in a computation, Spark will regenerate it automatically using the stored dependency graph.

val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val b = a.map(x => (x.length, x))
b.values.collect

zip

Joins two RDDs by combining the i-th of either partition with each other. The resulting RDD will consist of two-component tuples which are interpreted as key-value pairs by the methods provided by the PairRDDFunctions extension.
val a = sc.parallelize(1 to 100, 3)
val b = sc.parallelize(101 to 200, 3)
a.zip(b).collect

val a = sc.parallelize(1 to 100, 3)
val b = sc.parallelize(101 to 200, 3)
val c = sc.parallelize(201 to 300, 3)
a.zip(b).zip(c).map((x) => (x._1._1, x._1._2, x._2 )).collect

zipWithIndex

Zips the elements of the RDD with its element indexes. The indexes start from 0. If the RDD is spread across multiple partitions then a spark Job is started to perform this operation.
val z = sc.parallelize(Array("A", "B", "C", "D"))
val r = z.zipWithIndex
zipWithUniqueId

This is different from zipWithIndex since just gives a unique id to each data element but the ids may not match the index number of the data element. This operation does not start a spark job even if the RDD is spread across multiple partitions.
val z = sc.parallelize(100 to 120, 5)
val r = z.zipWithUniqueId
r.collect

Read full article from Apache Spark RDD API Examples

Parallel Programming With Spark

Parallel Programming With Spark
Improves efficiency through:
» In4memory computing primitives
»General computation graphs

Spark originally written in Scala, which allows
concise function syntax and interactive use
Recently added Java API for standalone apps

Concept: resilient distributed datasets (RDDs)
» Immutable collections of objects spread across a cluster
» Built through parallel transformations (map, filter, etc)
» Automatically rebuilt on failure
» Controllable persistence (e.g. caching in RAM) for reuse

Main Primitives
Resilient distributed datasets (RDDs)
» Immutable, partitioned collections of objects
Transformations (e.g. map, filter, groupBy, join)
» Lazy operations to build RDDs from other RDDs
Actions (e.g. count, collect, save)
»Return a result or write it to storage

RDD Fault Tolerance
RDDs track the series of transformations used to
build them (their lineage) to recompute lost data

Easiest way: Spark interpreter (spark-shell)
» Modified version of Scala interpreter for cluster use
Runs in local mode on 1 thread by default, but
can control through MASTER environment var:
MASTER=local ./spark-shell # local, 1 thread
MASTER=local[2] ./spark-shell # local, 2 threads
MASTER=host:port ./spark-shell # run on Mesos

First Stop: SparkContext
Main entry point to Spark functionality
Created for you in spark-shell as variable sc

Creating RDDs
// Turn a Scala collection into an RDD
sc.parallelize(List(1, 2, 3))
// Load text file from local FS, HDFS, or S3
sc.textFile("file.txt")
sc.textFile("directory/*.txt")
sc.textFile("hdfs://namenode:9000/path/file")
// Use any existing Hadoop InputFormat
sc.hadoopFile(keyClass, valClass, inputFmt, conf)

val nums = sc.parallelize(List(1, 2, 3))
// Pass each element through a function
val squares = nums.map(x => x*x) // {1, 4, 9}
// Keep elements passing a predicate
val even = squares.filter(_ % 2 == 0) // {4}
// Map each element to zero or more others
nums.flatMap(x => 1 to x) // => {1, 1, 2, 1, 2, 3}

Working with KeyUValue Pairs
Spark’s "distributed reduce" transformations
operate on RDDs of key4value pairs
Scala pair syntax:
val pair = (a, b) // sugar for new Tuple2(a, b)
Accessing pair elements:
pair._1 // => a
pair._2 // => b

Some KeyUValue Operations
val pets = sc.parallelize( List(("cat", 1), ("dog", 1), ("cat", 2)))
pets.reduceByKey(_ + _) // => {(cat, 3), (dog, 1)}
pets.groupByKey() // => {(cat, Seq(1, 2)), (dog, Seq(1)}
pets.sortByKey() // => {(cat, 1), (cat, 2), (dog, 1)}
reduceByKey also automatically implements
combiners on the map side

val lines = sc.textFile("hamlet.txt")
val counts = lines.flatMap(line => line.split(" "))
 .map(word => (word, 1))
 .reduceByKey(_ + _)

 val visits = sc.parallelize(List(
 ("index.html"
, "1.2.3.4"),
 ("about.html"
, "3.4.5.6"),
 ("index.html"
, "1.3.3.1")))
val pageNames = sc.parallelize(List(
 ("index.html"
, "Home"), ("about.html"
, "About")))
visits.join(pageNames)
// ("index.html", ("1.2.3.4", "Home"))
// ("index.html", ("1.3.3.1", "Home"))
// ("about.html", ("3.4.5.6", "About"))
visits.cogroup(pageNames)
// ("index.html", (Seq("1.2.3.4", "1.3.3.1"), Seq("Home")))
// ("about.html", (Seq("3.4.5.6"), Seq("About")))

Controlling The Number of
Reduce Tasks
All the pair RDD operations take an optional
second parameter for number of tasks
words.reduceByKey(_ + _, 5)
words.groupByKey(5)
visits.join(pageViews, 5)
Can also set spark.default.parallelism property

Using Local Variables
Any external variables you use in a closure will
automatically be shipped to the cluster:
val query = Console.readLine()
pages.filter(_.contains(query)).count()
Some caveats:
» Each task gets a new copy (updates aren’t sent back)
»Variable must be Serializable
»Don’t use fields of an outer object (ships all of it!)

sample(): deterministically sample a subset
union(): merge two RDDs
cartesian(): cross product
pipe(): pass through external program

Task Scheduler
Runs general task
graphs
Pipelines functions
where possible
Cache4aware data
reuse   locality
Partitioning4aware
to avoid shuffles
Please read full article from Parallel Programming With Spark

Histogram inScala | Big Data Analytics with Spark

Histogram inScala | Big Data Analytics with Spark
val hist=Array("aa","bb","aa").foldLeft(Map[String,Int]()){
 (m,c) => m + (c -> (m.getOrElse(c,0)+1))
}

val hist=Array("aa","bb","aa").foldLeft(Map[String,Int]()){
 (m,c) => m.updated(c, m.getOrElse(c,0)+1)
}

hist.toSeq.sortBy(_._1).foreach(println)
Read full article from Histogram inScala | Big Data Analytics with Spark

Histogram in Spark (1) | Big Data Analytics with Spark

Histogram in Spark (1) | Big Data Analytics with Spark

Spark’s DoubleRDDFunctions provide a histogram function for RDD[Double]. However there are no histogram function for RDD[String]. Here is a quick exercise for doing it. We will use immutable Map in this exercise.

Create a dummy RDD[String] and apply the aggregate method to calculate histogram

val d=sc.parallelize((1 to 10).map(_ % 3).map("val"+_.toString))

d.aggregate(Map[String,Int]())(
     (m,c)=>m.updated(c,m.getOrElse(c,0)+1),
     (m,n)=>(m /: n){case (map,(k,v))=>map.updated(k,v+map.getOrElse(k,0))}
)

def mapadd[T](m:Map[T,Int],n:Map[T,Int])={
     (m /: n){case (map,(k,v))=>map.updated(k,v+map.getOrElse(k,0))}
}

mapadd(Map("a"->1,"b"->2),Map("a"->2,"c"->1))

d.aggregate(Map[String,Int]())(
     (m,c)=>m.updated(c,m.getOrElse(c,0)+1),
     mapadd(_,_)
)
Read full article from Histogram in Spark (1) | Big Data Analytics with Spark

Spark 1.1.0 Basic Statistics（上） | 程序员的自我修养 | 关注Java、大数据、机器学习

Spark 1.1.0 Basic Statistics（上） | 程序员的自我修养 | 关注Java、大数据、机器学习

Summary statistics主要提供基于列的统计信息，包括6个统计量：均值、方差、非零统计量个数、总数、最小值、最大值。

SparkConf sparkConf = new SparkConf().setAppName("Statistics").setMaster("local[2]");

    JavaSparkContext sc = new JavaSparkContext(sparkConf);

    JavaRDD<String> data = sc.textFile("/home/yurnom/data/statistics.txt");

    JavaRDD<Vector> parsedData = data.map(s -> {

        double[] values = Arrays.asList(SPACE.split(s))

              .stream()

              .mapToDouble(Double::parseDouble)

              .toArray();

        return Vectors.dense(values);

    });

 

    MultivariateStatisticalSummary summary = Statistics.colStats(parsedData.rdd());

    System.out.println("均值:"+summary.mean());

    System.out.println("方差:"+summary.variance());

    System.out.println("非零统计量个数:"+summary.numNonzeros());

    System.out.println("总数:"+summary.count());

    System.out.println("最大值:"+summary.max());

    System.out.println("最小值:"+summary.min());

运行结果的含义对照测试数据来看是一目了然的，值得一提的是方差的计算方式为：

s2=∑ni=1(xi−x¯)2n−1

Correlations，相关度量，目前Spark支持两种相关性系数：皮尔逊相关系数（pearson）和斯皮尔曼等级相关系数（spearman）。相关系数是用以反映变量之间相关关系密切程度的统计指标。简单的来说就是相关系数绝对值越大（值越接近1或者-1）则表示数据越可进行线性拟合。如下图所示：

一个根据Key来抽样的功能，可以为每个key设置其被选中的概率。具体见代码以及注释。

Read full article from Spark 1.1.0 Basic Statistics（上） | 程序员的自我修养 | 关注Java、大数据、机器学习