scala RDD的example-国外大牛制作_rdd-exanple

转一个国外大牛的RDD的example：

连接地址是：http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

Our research group has a very strong focus on using and improving Apache Spark to solve real world programs. In order to do this we need to have a very solid understanding of the capabilities of Spark. So one of the first things we have done is to go through the entire Spark RDD API and write examples to test their functionality. This has been a very useful exercise and we would like to share the examples with everyone.

Authors of examples: Matthias Langer and Zhen He
Emails addresses: m.langer@latrobe.edu.au, z.he@latrobe.edu.au

These examples have only been tested for Spark version 1.1. We assume the functionality of Spark is stable and therefore the examples should be valid for later releases.


The RDD API By Example

RDD is short for Resilient Distributed Dataset. RDDs are the workhorse of the Spark system. As a user, one can consider a RDD as a handle for a collection of individual data partitions, which are the result of some computation.

However, 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.

One reason why Spark has lately become a very popular system for processing big data is that it does not impose restrictions regarding what data can be stored within RDD partitions. The RDD API already contains many useful operations. But, because the creators of Spark had to keep the core API of RDDs common enough to handle arbitrary data-types, many convenience functions are missing.

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

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- nent key-value tuples as required by the PairRDDFunctions. However, there are additional requirements considering the convertibility of the tuple components to Writable types.

Since Spark will make methods with extended functionality automatically available to users when the data items fulfill the above described requirements, we decided to list all possible available functions in strictly alphabetical order. We will append either of the followingto the function-name to indicate it belongs to an extension that requires the data items to conform to a certain format or type.

[Double] - Double RDD Functions

[Ordered] - OrderedRDDFunctions

[Pair] - PairRDDFunctions

[SeqFile] - SequenceFileRDDFunctions

---

aggregate

The aggregate-method provides an interface for performing highly customized reductions and aggregations with a RDD. However, due to the way Scala and Spark execute and process data, care must be taken to achieve deterministic behavior. The following list contains a few observations we made while experimenting with aggregate:

• The reduce and combine functions have to be commutative and associative.
• As can be seen from the function definition below, the output of the combiner must be equal to its input. This is necessary because Spark will chain-execute it.
• The zero value is the initial value of the U component when either seqOp or combOp are executed for the first element of their domain of influence. Depending on what you want to achieve, you may have to change it. However, to make your code deterministic, make sure that your code will yield the same result regardless of the number or size of partitions.
• Do not assume any execution order for either partition computations or combining partitions.
• The neutral zeroValue is applied at the beginning of each sequence of reduces within the individual partitions and again when the output of separate partitions is combined.
• Why have two separate combine functions? The first functions maps the input values into the result space. Note that the aggregation data type (1st input and output) can be different (U != T). The second function reduces these mapped values in the result space.
• Why would one want to use two input data types? Let us assume we do an archaeological site survey using a metal detector. While walking through the site we take GPS coordinates of important findings based on the output of the metal detector. Later, we intend to draw an image of a map that highlights these locations using the aggregate function. In this case the zeroValue could be an area map with no highlights. The possibly huge set of input data is stored as GPS coordinates across many partitions. seqOp could convert the GPS coordinates to map coordinates and put a marker on the map at the respective position. combOp will receive these highlights as partial maps and combine them into a single final output map.

Listing Variants

Examples 1

The main issue with the code above is that the result of the inner min is a string of length 1.
The zero in the output is due to the empty string being the last string in the list. We see this result because we are not recursively reducing any further within the partition for the final string.

Examples 2

In contrast to the previous example, this example has the empty string at the beginning of the second partition. This results in length of zero being input to the second reduce which then upgrades it a length of 1. (Warning: The above example shows bad design since the output is dependent on the order of the data inside the partitions.)

---

cartesian

Listing Variants

Example

---

checkpoint

Will create a checkpoint when the RDD is computed next. Checkpointed RDDs are stored as a binary file within the checkpoint directory which can be specified using the Spark context. (Warning: Spark applies lazy evaluation. Checkpointing will not occur until an action is invoked.)

Important note: the directory  "my_directory_name" should exist in all slaves. As an alternative you could use an HDFS directory URL as well.

Listing Variants

Example

---

coalesce, repartition

Listing Variants


Example

---

cogroup [Pair], groupWith [Pair]

Listing Variants


Examples

---

collect, toArray

Listing Variants


Example

---

collectAsMap [Pair]

Listing Variants


Example

---

combineByKey[Pair]

Listing Variants

Example

---

compute

Listing Variants

---

context, sparkContext

Listing Variants


Example

---

count

Listing Variants


Example

---

countApprox

Marked as experimental feature! Experimental features are currently not covered by this document! Listing Variants

---

countByKey [Pair]

Very similar to count, but counts the values of a RDD consisting of two-component tuples for each distinct key separately. Listing Variants


Example

---

countByKeyApprox [Pair]

Marked as experimental feature! Experimental features are currently not covered by this document! Listing Variants

---

countByValue

Returns a map that contains all unique values of the RDD and their respective occurrence counts. (Warning: This operation will finally aggregate the information in a single reducer.)

Listing Variants


Example

---

countByValueApprox

Marked as experimental feature! Experimental features are currently not covered by this document! Listing Variants

---

countApproxDistinct

Computes the approximate number of distinct values. For large RDDs which are spread across many nodes, this function may execute faster than other counting methods. The parameter relativeSD controls the accuracy of the computation.

Listing Variants


Example

---

countApproxDistinctByKey [Pair]
 
Similar to countApproxDistinct, but computes the approximate number of distinct values for each distinct key. Hence, the RDD must consist of two-component tuples. For large RDDs which are spread across many nodes, this function may execute faster than other counting methods. The parameter relativeSD controls the accuracy of the computation.

Listing Variants


Example

---

dependencies
 
Returns the RDD on which this RDD depends.

Listing Variants


Example

---

distinct
 
Returns a new RDD that contains each unique value only once.

Listing Variants


Example

---

first
 
Looks for the very first data item of the RDD and returns it.

Listing Variants


Example

---

filter
 
Evaluates a boolean function for each data item of the RDD and puts the items for which the function returned true into the resulting RDD.

Listing Variants


Example


When you provide a filter function, it must be able to handle all data items contained in the RDD. Scala provides so-called partial functions to deal with mixed data-types. (Tip: Partial functions are very useful if you have some data which may be bad and you do not want to handle but for the good data (matching data) you want to apply some kind of map function. The following article is good. It teaches you about partial functions in a very nice way and explains why case has to be used for partial functions:   article)

Examples for mixed data without partial functions


This fails because some components of a are not implicitly comparable against integers. Collect uses the isDefinedAt property of a function-object to determine whether the test-function is compatible with each data item. Only data items that pass this test (=filter) are then mapped using the function-object.

Examples for mixed data with partial functions



Be careful! The above code works because it only checks the type itself! If you use operations on this type, you have to explicitly declare what type you want instead of any. Otherwise the compiler does (apparently) not know what bytecode it should produce:

---

filterWith
 
This is an extended version of filter. It takes two function arguments. The first argument must conform to Int -> T and is executed once per partition. It will transform the partition index to type T. The second function looks like (U, T) -> Boolean. T is the transformed partition index and U are the data items from the RDD. Finally the function has to return either true or false (i.e. Apply the filter).

Listing Variants


Example

---

flatMap
 
Similar to map, but allows emitting more than one item in the map function.

Listing Variants


Example

---

flatMapValues
 
Very similar to mapValues, but collapses the inherent structure of the values during mapping.

Listing Variants


Example

---

flatMapWith
 
Similar to flatMap, but allows accessing the partition index or a derivative of the partition index from within the flatMap-function.

Listing Variants


Example

---

fold
 
Aggregates the values of each partition. The aggregation variable within each partition is initialized with zeroValue.

Listing Variants


Example

---

foldByKey [Pair]
 
Very similar to fold, but performs the folding separately for each key of the RDD. This function is only available if the RDD consists of two-component tuples.

Listing Variants


Example

---

foreach
 
Executes an parameterless function for each data item.

Listing Variants


Example

---

foreachPartition
 
Executes an parameterless function for each partition. Access to the data items contained in the partition is provided via the iterator argument.

Listing Variants


Example

---

foreachWith
 
Executes an parameterless function for each partition. Access to the data items contained in the partition is provided via the iterator argument.

Listing Variants


Example

---

generator, setGenerator
 
Allows setting a string that is attached to the end of the RDD's name when printing the dependency graph.

Listing Variants

---

getCheckpointFile
 
Returns the path to the checkpoint file or null if RDD has not yet been checkpointed.

Listing Variants


Example

---

preferredLocations
 
Returns the hosts which are preferred by this RDD. The actual preference of a specific host depends on various assumptions.

Listing Variants

---

getStorageLevel
 
Retrieves the currently set storage level of the RDD. This can only be used to assign a new storage level if the RDD does not have a storage level set yet. The example below shows the error you will get, when you try to reassign the storage level.

Listing Variants


Example

---

glom
 
Assembles an array that contains all elements of the partition and embeds it in an RDD.

Listing Variants


Example

---

groupBy
 

Listing Variants


Example

---

groupByKey [Pair]
 
Very similar to groupBy, but instead of supplying a function, the key-component of each pair will automatically be presented to the partitioner.

Listing Variants


Example

---

histogram [Double]
 
These functions take an RDD of doubles and create a histogram with either even spacing (the number of buckets equals to bucketCount) or arbitrary spacing based on  custom bucket boundaries supplied by the user via an array of double values. The result type of both variants is slightly different, the first function will return a tuple consisting of two arrays. The first array contains the computed bucket boundary values and the second array contains the corresponding count of values (i.e. the histogram). The second variant of the function will just return the histogram as an array of integers.

Listing Variants


Example with even spacing



Example with custom spacing

---

id

Retrieves the ID which has been assigned to the RDD by its device context.

Listing Variants


Example

---

intersection

Returns the elements in the two RDDs which are the same.

Listing Variants


Example

---

isCheckpointed

Indicates whether the RDD has been checkpointed. The flag will only raise once the checkpoint has really been created.

Listing Variants


Example

---

iterator

Returns a compatible iterator object for a partition of this RDD. This function should never be called directly.

Listing Variants

---

join [Pair]

Performs an inner join using two key-value RDDs. Please note that the keys must be generally comparable to make this work.

Listing Variants


Example

---

keyBy

Constructs two-component tuples (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.

Listing Variants


Example

---

keys [Pair]

Extracts the keys from all contained tuples and returns them in a new RDD.

Listing Variants


Example

---

leftOuterJoin [Pair]

Performs an left outer join using two key-value RDDs. Please note that the keys must be generally comparable to make this work correctly.

Listing Variants


Example

---

lookup

Scans the RDD for all keys that match the provided value and returns their values as a Scala sequence.


Listing Variants


Example

---

map

Applies a transformation function on each item of the RDD and returns the result as a new RDD.


Listing Variants


Example

---

mapPartitions

This is a specialized map that is called only once for each partition. The entire content of the respective partitions is available as a sequential stream of values via the input argument ( Iterarator[T]). The custom function must return yet another Iterator[U]. The combined result iterators are automatically converted into a new RDD. Please note, that the tuples (3,4) and (6,7) are missing from the following result due to the partitioning we chose.


Listing Variants


Example 1


Example 2


The above program can also be written using flatMap as follows.

Example 2 using flatmap

---

mapPartitionsWithContext   (developer API)

Similar to mapPartitions, but allows accessing information about the processing state within the mapper.


Listing Variants


Example

---

mapPartitionsWithIndex

Similar to mapPartitions, but takes two parameters. The first parameter is the index of the partition and the second is an iterator through all the items within this partition. The output is an iterator containing the list of items after applying whatever transformation the function encodes.


Listing Variants


Example

---

mapPartitionsWithSplit

This method has been marked as deprecated in the API. So, you should not use this method anymore. Deprecated methods will not be covered in this document.


Listing Variants

---

mapValues [Pair]

Takes the values of a RDD that consists of two-component tuples, and applies the provided function to transform each value. Then, it forms new two-component tuples using the key and the transformed value and stores them in a new RDD.


Listing Variants


Example

---

mapWith  (deprecated)

This is an extended version of map. It takes two function arguments. The first argument must conform to Int -> T and is executed once per partition. It will map the partition index to some transformed partition index of type T. This is where it is nice to do some kind of initialization code once per partition. Like create a Random number generator object. The second function must conform to (U, T) -> U. T is the transformed partition index and U is a data item of the RDD. Finally the function has to return a transformed data item of type U.


Listing Variants


Example

---

max

Returns the largest element in the RDD

Listing Variants


Example

---

mean [Double], meanApprox [Double]

Calls stats and extracts the mean component. The approximate version of the function can finish somewhat faster in some scenarios. However, it trades accuracy for speed.


Listing Variants


Example

---

min

Returns the smallest element in the RDD

Listing Variants


Example

---

name, setName

Allows a RDD to be tagged with a custom name.

Listing Variants


Example

---

partitionBy [Pair]

Repartitions as key-value RDD using its keys. The partitioner implementation can be supplied as the first argument.

Listing Variants

---

partitioner

Specifies a function pointer to the default partitioner that will be used for groupBy, subtract, reduceByKey (from PairedRDDFunctions), etc. functions.

Listing Variants

---

partitions

Returns an array of the partition objects associated with this RDD.

Listing Variants



Example

---

persist, cache

These functions can be used to adjust the storage level of a RDD. When freeing up memory, Spark will use the storage level identifier to decide which partitions should be kept. The parameterless variants persist() and cache() are just abbreviations for persist(StorageLevel.MEMORY_ONLY). (Warning: Once the storage level has been changed, it cannot be changed again!)

Listing Variants



Example

---

pipe

Takes the RDD data of each partition and sends it via stdin to a shell-command. The resulting output of the command is captured and returned as a RDD of string values.

Listing Variants



Example

---

randomSplit

Randomly splits an RDD into multiple smaller RDDs according to a weights Array which specifies the percentage of the total data elements that is assigned to each smaller RDD. Note the actual size of each smaller RDD is only approximately equal to the percentages specified by the weights Array. The second example below shows the number of items in each smaller RDD does not exactly match the weights Array.   A random optional seed can be specified. This function is useful for spliting data into a training set and a testing set for machine learning.

Listing Variants



Example

---

reduce

This function provides the well-known reduce functionality in Spark. Please note that any function f you provide, should be commutative in order to generate reproducible results.

Listing Variants



Example

---

reduceByKey [Pair],  reduceByKeyLocally [Pair], reduceByKeyToDriver [Pair]

This function provides the well-known reduce functionality in Spark. Please note that any function f you provide, should be commutative in order to generate reproducible results.

Listing Variants



Example

---

rightOuterJoin [Pair]

Performs an right outer join using two key-value RDDs. Please note that the keys must be generally comparable to make this work correctly.

Listing Variants



Example

---

sample

Randomly selects a fraction of the items of a RDD and returns them in a new RDD.

Listing Variants



Example

---

saveAsHadoopFile [Pair], saveAsHadoopDataset [Pair], saveAsNewAPIHadoopFile [Pair]

Saves the RDD in a Hadoop compatible format using any Hadoop outputFormat class the user specifies.

Listing Variants

---

saveAsObjectFile

Saves the RDD in binary format.

Listing Variants


Example

---

saveAsSequenceFile [SeqFile]

Saves the RDD as a Hadoop sequence file.

Listing Variants


Example

---

saveAsTextFile

Saves the RDD as text files. One line at a time.

Listing Variants


Example without compression



Example with compression



Example writing into HDFS

---

stats [Double]

Simultaneously computes the mean, variance and the standard deviation of all values in the RDD.

Listing Variants


Example

---

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.

Listing Variants


Example

---

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.

Listing Variants


Example

---

stdev [Double], sampleStdev [Double]

Calls stats and extracts either stdev-component or corrected sampleStdev-component.

Listing Variants


Example

---

subtract

Performs the well known standard set subtraction operation: A - B

Listing Variants


Example

---

subtractByKey [Pair]

Very similar to subtract, but instead of supplying a function, the key-component of each pair will be automatically used as criterion for removing items from the first RDD.

Listing Variants


Example

---

sum [Double], sumApprox [Double]

Computes the sum of all values contained in the RDD. The approximate version of the function can finish somewhat faster in some scenarios. However, it trades accuracy for speed.


Listing Variants


Example

---

take

Extracts the first n items of the RDD and returns them as an array. (Note: This sounds very easy, but it is actually quite a tricky problem for the implementors of Spark because the items in question can be in many different partitions.)


Listing Variants


Example

---

takeOrdered

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


Listing Variants


Example

---

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.

Listing Variants


Example

---

toDebugString

Returns a string that contains debug information about the RDD and its dependencies.


Listing Variants


Example

---

toJavaRDD

Embeds this RDD object within a JavaRDD object and returns it.


Listing Variants


Example

---

top

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

Listing Variants


Example

---

toString

Assembles a human-readable textual description of the RDD.


Listing Variants


Example

---

union, ++

Performs the standard set operation: A union B


Listing Variants


Example

---

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.


Listing Variants


Example

---

values

Extracts the values from all contained tuples and returns them in a new RDD.


Listing Variants


Example

---

variance [Double], sampleVariance [Double]

Calls stats and extracts either variance-component or corrected sampleVariance-component.

Listing Variants


Example

---

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.

Listing Variants


Example

---

zipParititions

Similar to zip. But provides more control over the zipping process.

Listing Variants


Example

---

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.

Listing Variants


Example

---

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.
Compare the results of the example below with that of the 2nd example of zipWithIndex. You should be able to see the difference.

Listing Variants


Example