SparkContext.parallelize is mainly used to learn Spark in the Spark shell, since it requires all the data to be available on the Spark driver.
An RDD is a named (by name) and uniquely identified (by id) entity inside a SparkContext.
When an RDD is created, it belongs to and is completely owned by the Spark context it originated from. RDDs can’t be shared between SparkContexts, by design
\n" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "status": "READY", + "errorMessage": "", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:210" + }, + { + "text": "// You can optionally assign custom names to RDDs\n\nval ns = sc.parallelize(0 to 10)\nns.name = \"foo\"\nns.toDebugString", + "dateUpdated": "2016-06-13T16:25:20+0000", + "config": { + "enabled": true, + "tableHide": true, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "editorMode": "ace/mode/scala", + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120969_1010188776", + "id": "20160524-190150_592895773", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "ns: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[12] at parallelize atNamed RDDs are also easily tracked in the Web UI
\n" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "status": "READY", + "errorMessage": "", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:212" + }, + { + "text": "val ints = sc.parallelize(1 to 100) \n ints.setName(\"Hundred ints\") \n ints.cache \n ints.count", + "dateUpdated": "2016-06-13T16:25:20+0000", + "config": { + "enabled": true, + "tableHide": true, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "editorMode": "ace/mode/scala", + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120969_1010188776", + "id": "20160524-190322_141239459", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "ints: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[13] at parallelize atOnce created, RDDs offer two types of operations: transformations and actions.
\n- \n
- Transformations are lazy operations that construct a new RDD from a previous one. \n
- Actions are operations that trigger computation and return (non-RDD) values. \n
Element-wise transformations
\nThe two most common transformations you will likely be using are map and filter. A snippet from lab 1 illustrates this:
val notFoundErrors = accessLogsClean\n.map(log => (log.ipAddress, log.responseCode))\n.filter(_._2 == 404)\nPair RDDs
\nSpark provides special operations on RDDs containing key/value pairs. These RDDs are called pair RDDs and have a number of PairRDDFunctions that can be used to manipulate them.
Pair RDDs are a useful building block in many programs, as they expose operations that allow you to act on each key in parallel or regroup data\n
across the network.
Pair-generating transformations
\nThe following transformations generate RDDs of tuples. The resulting RDD will consist of two-component tuples which can be interpreted as key-value pairs.
\nZip
\nZip joins two RDDs by combining the i-th of either partition with each other. This is analogous to the zip operation on Scala collections with the additional requirement that both RDDs are partitioned identically. That is, the RDDs must have the same number of partitions and the same number of elements in each partition.
KeyBy
\nkeyBy constructs key-value pairs by applying a function on each data item. The result of the function becomes the key and the original data item becomes the value of the newly created tuples.
GroupBy
\ngroupBy constructs key-value pairs by applying a function on each data item and grouping the results.
Pair-RDD transformations
\nA number of methods have cousins with names like fooByKey or fooValues. These cousins perform the analagous operations on the first or second part of each tuple respectively.
Note that the keys are carried along with each new value created.
\nFlatMapValues
\nflatMapValues runs each value in the key-value pair RDD through a flatMap function without changing the keys.
GroupByKey
\ngroupByKey is similar to groupBy, but with the key-component of each pair as the grouping function.
combineByKey combines the values of a RDD consisting of two-component tuples by applying multiple aggregators one after another.
combineByKey is the most general of the per-key aggregation functions. Most of the other per-key combiners are implemented using it.
Like aggregate, combineByKey allows the user to return values that are not the same type as our input data.
As combineByKey goes through the elements in a partition, each element either has a key it hasn’t seen before or has the same key as a previous element.
If it is a value we have seen before while processing that partition, it will instead use the second provided function (g in the above example) with the current value for the accumulator for that key and the new value.\n
Since each partition is processed independently, we can have multiple accumulators for the same key. When we are merging the results from each partition, if two or more partitions have an accumulator for the same key we merge the accumulators using the third provided function (h in the above example).
We can disable map-side aggregation in combineByKey if we know that our data won’t benefit from it. For example, groupByKey disables map-side aggregation as the aggregation function (appending to a list) does not save any space.
\nIf it’s a new element, combineByKey uses a function we provide (f in the above example) to create the initial value for the accumulator on that key.
Sorting Data
\nWe can sort an RDD with key/value pairs provided that there is an ordering defined on the key.
\nsortByKey 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 (more on shuffled RDDs soon).
The implementation of sortByKey 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.
The implementation of sortByKey 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.
SQL transformations
\nJoin
\njoin performs an inner join using two pair RDDs. The keys must generally be of the same type.
![\"innerjoin\"](\"http://www.vertabelo.com/_file/blog/sql-joins/sql-joins-venn-diagrams-inner-join.png\")
Left Outer Join
\nleftOuterJoin performs a left outer join using two key-value RDDs.
![\"loj\"](\"http://www.vertabelo.com/_file/blog/sql-joins/sql-joins-venn-diagrams-left-outer-join.png\")
Full Outer Join
\nfullOuterJoin performs the full outer join between two pair RDDs.
![\"foj\"](\"http://www.vertabelo.com/_file/blog/sql-joins/sql-joins-venn-diagrams-full-outer-join.png\")
Co-group
\ncogroup groups data from both RDDs by key.
Actions
\nThe simplest and most common action that returns data to our driver program is collect, which returns the entire RDD’s contents.
collect suffers from the restriction that all of your data must fit on a single machine, as it all needs to be copied to the driver.
Fold
\nfold aggregates the values of each partition. The aggregation variable within each partition is initialized with zeroValue, which should be the identity element for your operation.
Note the difference between foldLeft and foldRight.
Reduce
\nThe most common action on basic RDDs you will likely use is reduce, which takes a function that operates on two elements of the type in your RDD and returns a new element of the same type.
This function provides the well-known reduce functionality in Spark. Please note that any function you provide should be commutative in order to generate deterministic results.
\n" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "status": "READY", + "errorMessage": "", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:245" + }, + { + "text": "//def reduce(f: (T, T) => T): T\nval a = sc.parallelize(1 to 100, 3)\na.reduce(_ + _)", + "dateUpdated": "2016-06-13T16:25:20+0000", + "config": { + "enabled": true, + "tableHide": false, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120974_1009804027", + "id": "20160601-191837_910151641", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[40] at parallelize atBoth fold and reduce require that the return type of our result be the same type as that of the elements in the RDD we are operating over.
This works well for operations like sum, but sometimes we want to return a different type. For example, when computing a running average, we need to keep track of both the count so far and the number of elements, which requires us to return a pair.
\nAggregate
\nWe can use aggregate to compute the average of an RDD, avoiding a map before the fold.
Pair RDD Actions
\nThe following operations are actions on PairRDDs.
Lookup
\nlookup scans the RDD for all keys that match the provided value and returns their values as a Scala sequence.
CountByKey
\ncountByKey is very similar to count, but it counts the values of a PairRDD consisting of two-component tuples for each distinct key seperately.
CollectAsMap
\ncollectAsMap is similar to collect but it converts a PairRDD into a Scala Map which preserves the key-value structure. See also the combineByKey example above.
FoldByKey
\nfoldByKey is analogous to fold, but it performs the fold operation separately for each key of the PairRDD.
ReduceByKey
\nreduceByKey provides the well-known reduce operation in Spark. Any function provided as an argument to reduceByKey must be commutative in order to generate deterministic results.
Problems
\nFill in your answers in the empty cell below each question.
\n" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "status": "READY", + "errorMessage": "", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:259" + }, + { + "text": "val z = sc.parallelize(List(1,2,3,4,5,6), 2)\n\n// lets first print out the contents of the RDD with partition labels\ndef foo(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {\n iter.toList.map(x => \"[partID:\" + index + \", val: \" + x + \"]\").iterator\n}\n\nz.mapPartitionsWithIndex(foo).collect", + "dateUpdated": "2016-06-13T16:25:20+0000", + "config": { + "enabled": true, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120975_1009419278", + "id": "20160602-171519_1652857423", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "z: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[59] at parallelize atProblem 1
\nRecall the function signature for aggregate.
aggregate[U](zeroValue: U)(intra: (U, T) => U, inter: (U, U) => U): U\nCompute the sum of the maximal values in each partition. What happens if you pass in List(1, 2, 3, 4, 5, 6).max + 1 as the zero value?
Problem 2
\nUse aggregate to concatenate List(\"a\",\"b\",\"c\",\"d\",\"e\",\"f\") parallelized over two partitions.
Problem 3
\nLet's look at some actual data. This data captures two metrics: the monthly total for revolving credit owned by commercial institutions and the monthly rate of flow for revolving credit owned by commercial instuitions. Both metrics are in millions of dollars, and neither is seasonally adjusted. The data is stored as a CSV file on S3.
\nPart 1
\nThe first two lines of the data have information that is not of interest. Write a function drop(rdd: RDD[T], n: Int): RDD[T] that drops the first n lines of data from the file. Note that RDDs take a type parameter, and are not covariant. You can specify an explicit type for T in your function definition.
Part 2
\nWhat is the average monthly revolving credit in the dataset?
\nPart 3
\nWhat are the top 3 months with the highest absolute revolving credit? What are the months with the highest positive gain in revolving credit? How about the months with the highest overall net change in revolving credit?
\n" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "status": "FINISHED", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:265", + "authenticationInfo": {}, + "dateFinished": "2016-06-13T18:38:04+0000", + "dateStarted": "2016-06-13T18:38:04+0000", + "focus": true + }, + { + "text": "// Solution Part 1", + "authenticationInfo": {}, + "dateUpdated": "2016-06-13T16:26:04+0000", + "config": { + "enabled": true, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "editorMode": "ace/mode/scala", + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120976_1019807499", + "id": "20160610-214713_305233970", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "dateStarted": "2016-06-13T16:26:04+0000", + "dateFinished": "2016-06-13T16:26:05+0000", + "status": "FINISHED", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:266" + }, + { + "text": "// Solution Part 2", + "authenticationInfo": {}, + "dateUpdated": "2016-06-13T16:26:08+0000", + "config": { + "enabled": true, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "editorMode": "ace/mode/scala", + "colWidth": 12 + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835120976_1019807499", + "id": "20160610-215253_1503804127", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "" + }, + "dateCreated": "2016-06-13T16:25:20+0000", + "dateStarted": "2016-06-13T16:26:08+0000", + "dateFinished": "2016-06-13T16:26:08+0000", + "status": "FINISHED", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:267" + }, + { + "text": "// Solution Part 3", + "authenticationInfo": {}, + "dateUpdated": "2016-06-13T16:26:20+0000", + "config": { + "colWidth": 12, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "enabled": true, + "editorMode": "ace/mode/scala" + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835168426_1443308769", + "id": "20160613-162608_486807350", + "result": { + "code": "SUCCESS", + "type": "TEXT", + "msg": "" + }, + "dateCreated": "2016-06-13T16:26:08+0000", + "dateStarted": "2016-06-13T16:26:20+0000", + "dateFinished": "2016-06-13T16:26:21+0000", + "status": "FINISHED", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:268" + }, + { + "config": { + "colWidth": 12, + "graph": { + "mode": "table", + "height": 300, + "optionOpen": false, + "keys": [], + "values": [], + "groups": [], + "scatter": {} + }, + "enabled": true + }, + "settings": { + "params": {}, + "forms": {} + }, + "jobName": "paragraph_1465835180922_-101842814", + "id": "20160613-162620_1255253356", + "dateCreated": "2016-06-13T16:26:20+0000", + "status": "READY", + "progressUpdateIntervalMs": 500, + "$$hashKey": "object:269" + } + ], + "name": "RDD API Tutorial - No Solutions", + "id": "2BNQ1VVN6", + "angularObjects": { + "2BNFDK6C1:shared_process": [], + "2BMJXP95T:shared_process": [], + "2BP2Y3J9P:shared_process": [], + "2BPKQATHV:shared_process": [], + "2BKKRN36Z:shared_process": [], + "2BM55D463:shared_process": [], + "2BMTPQBYG:shared_process": [], + "2BKB9Q5F2:shared_process": [], + "2BNYK5NWH:shared_process": [], + "2BNSNR8J8:shared_process": [], + "2BPKH2W6R:shared_process": [], + "2BMTBRB17:shared_process": [], + "2BKX6W1Y6:shared_process": [], + "2BNQB5TA4:shared_process": [], + "2BM7Y5ZRK:shared_process": [], + "2BN675825:shared_process": [], + "2BNVQ5116:shared_process": [], + "2BMSC321X:shared_process": [], + "2BKA4KPBJ:shared_process": [], + "2BPHHAPHJ:shared_process": [] + }, + "config": { + "looknfeel": "default" + }, + "info": {} +} \ No newline at end of file