Heading
--> - - diff --git a/docs/_plugins/copy_api_dirs.rb b/docs/_plugins/copy_api_dirs.rb index acc6bf08160eb..44d64057f4fb3 100644 --- a/docs/_plugins/copy_api_dirs.rb +++ b/docs/_plugins/copy_api_dirs.rb @@ -20,7 +20,10 @@ if not (ENV['SKIP_API'] == '1' or ENV['SKIP_SCALADOC'] == '1') # Build Scaladoc for Java/Scala - projects = ["core", "examples", "repl", "bagel", "graphx", "streaming", "mllib"] + core_projects = ["core", "examples", "repl", "bagel", "graphx", "streaming", "mllib"] + external_projects = ["flume", "kafka", "mqtt", "twitter", "zeromq"] + + projects = core_projects + external_projects.map { |project_name| "external/" + project_name } puts "Moving to project root and building scaladoc." curr_dir = pwd diff --git a/docs/_plugins/production_tag.rb b/docs/_plugins/production_tag.rb new file mode 100644 index 0000000000000..9f870cf2137af --- /dev/null +++ b/docs/_plugins/production_tag.rb @@ -0,0 +1,14 @@ +module Jekyll + class ProductionTag < Liquid::Block + + def initialize(tag_name, markup, tokens) + super + end + + def render(context) + if ENV['PRODUCTION'] then super else "" end + end + end +end + +Liquid::Template.register_tag('production', Jekyll::ProductionTag) diff --git a/docs/bagel-programming-guide.md b/docs/bagel-programming-guide.md index cffa55ee952b0..da6d0c9dcd97b 100644 --- a/docs/bagel-programming-guide.md +++ b/docs/bagel-programming-guide.md @@ -16,7 +16,7 @@ This guide shows the programming model and features of Bagel by walking through To use Bagel in your program, add the following SBT or Maven dependency: groupId = org.apache.spark - artifactId = spark-bagel_{{site.SCALA_VERSION}} + artifactId = spark-bagel_{{site.SCALA_BINARY_VERSION}} version = {{site.SPARK_VERSION}} # Programming Model @@ -108,7 +108,7 @@ _Example_ ## Operations -Here are the actions and types in the Bagel API. See [Bagel.scala](https://github.com/apache/incubator-spark/blob/master/bagel/src/main/scala/org/apache/spark/bagel/Bagel.scala) for details. +Here are the actions and types in the Bagel API. See [Bagel.scala](https://github.com/apache/spark/blob/master/bagel/src/main/scala/org/apache/spark/bagel/Bagel.scala) for details. ### Actions diff --git a/docs/building-with-maven.md b/docs/building-with-maven.md index b9ff0af76f647..ded12926885b9 100644 --- a/docs/building-with-maven.md +++ b/docs/building-with-maven.md @@ -17,10 +17,10 @@ You'll need to configure Maven to use more memory than usual by setting `MAVEN_O If you don't run this, you may see errors like the following: - [INFO] Compiling 203 Scala sources and 9 Java sources to /Users/me/Development/spark/core/target/scala-{{site.SCALA_VERSION}}/classes... + [INFO] Compiling 203 Scala sources and 9 Java sources to /Users/me/Development/spark/core/target/scala-{{site.SCALA_BINARY_VERSION}}/classes... [ERROR] PermGen space -> [Help 1] - [INFO] Compiling 203 Scala sources and 9 Java sources to /Users/me/Development/spark/core/target/scala-{{site.SCALA_VERSION}}/classes... + [INFO] Compiling 203 Scala sources and 9 Java sources to /Users/me/Development/spark/core/target/scala-{{site.SCALA_BINARY_VERSION}}/classes... [ERROR] Java heap space -> [Help 1] You can fix this by setting the `MAVEN_OPTS` variable as discussed before. @@ -71,8 +71,8 @@ This setup works fine in IntelliJ IDEA 11.1.4. After opening the project via the ## Building Spark Debian Packages ## -It includes support for building a Debian package containing a 'fat-jar' which includes the repl, the examples and bagel. This can be created by specifying the following profiles: +The maven build includes support for building a Debian package containing the assembly 'fat-jar', PySpark, and the necessary scripts and configuration files. This can be created by specifying the following: - $ mvn -Prepl-bin -Pdeb clean package + $ mvn -Pdeb -DskipTests clean package -The debian package can then be found under repl/target. We added the short commit hash to the file name so that we can distinguish individual packages build for SNAPSHOT versions. +The debian package can then be found under assembly/target. We added the short commit hash to the file name so that we can distinguish individual packages built for SNAPSHOT versions. diff --git a/docs/configuration.md b/docs/configuration.md index da70cabba2d9b..3f03d97e8054c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -18,8 +18,8 @@ Spark provides three locations to configure the system: Spark properties control most application settings and are configured separately for each application. The preferred way to set them is by passing a [SparkConf](api/core/index.html#org.apache.spark.SparkConf) class to your SparkContext constructor. -Alternatively, Spark will also load them from Java system properties (for compatibility with old versions -of Spark) and from a [`spark.conf` file](#configuration-files) on your classpath. +Alternatively, Spark will also load them from Java system properties, for compatibility with old versions +of Spark. SparkConf lets you configure most of the common properties to initialize a cluster (e.g., master URL and application name), as well as arbitrary key-value pairs through the `set()` method. For example, we could @@ -98,7 +98,7 @@ Apart from these, the following properties are also available, and may be useful
groupByKey,
+ Default number of tasks to use across the cluster for distributed shuffle operations (groupByKey,
reduceByKey, etc) when not set by user.
BlockManager might take a performance hit.
+ 
+
+
GraphX optimizes the representation of `VD` and `ED` when they are plain old data-types (e.g.,
-> int, double, etc...) reducing the in memory footprint.
+> GraphX optimizes the representation of vertex and edge types when they are plain old data-types
+> (e.g., int, double, etc...) reducing the in memory footprint by storing them in specialized
+> arrays.
-In some cases we may wish to have vertices with different property types in the same graph. This can
-be accomplished through inheritance. For example to model users and products as a bipartite graph
-we might do the following:
+In some cases it may be desirable to have vertices with different property types in the same graph.
+This can be accomplished through inheritance. For example to model users and products as a
+bipartite graph we might do the following:
{% highlight scala %}
class VertexProperty()
@@ -116,9 +132,11 @@ var graph: Graph[VertexProperty, String] = null
{% endhighlight %}
Like RDDs, property graphs are immutable, distributed, and fault-tolerant. Changes to the values or
-structure of the graph are accomplished by producing a new graph with the desired changes. The graph
-is partitioned across the workers using a range of vertex-partitioning heuristics. As with RDDs,
-each partition of the graph can be recreated on a different machine in the event of a failure.
+structure of the graph are accomplished by producing a new graph with the desired changes. Note
+that substantial parts of the original graph (i.e., unaffected structure, attributes, and indicies)
+are reused in the new graph reducing the cost of this inherently functional data-structure. The
+graph is partitioned across the workers using a range of vertex-partitioning heuristics. As with
+RDDs, each partition of the graph can be recreated on a different machine in the event of a failure.
Logically the property graph corresponds to a pair of typed collections (RDDs) encoding the
properties for each vertex and edge. As a consequence, the graph class contains members to access
@@ -131,12 +149,12 @@ class Graph[VD, ED] {
}
{% endhighlight %}
-The classes `VertexRDD[VD]` and `EdgeRDD[ED]` extend and are optimized versions of `RDD[(VertexId,
+The classes `VertexRDD[VD]` and `EdgeRDD[ED]` extend and are optimized versions of `RDD[(VertexID,
VD)]` and `RDD[Edge[ED]]` respectively. Both `VertexRDD[VD]` and `EdgeRDD[ED]` provide additional
functionality built around graph computation and leverage internal optimizations. We discuss the
`VertexRDD` and `EdgeRDD` API in greater detail in the section on [vertex and edge
RDDs](#vertex_and_edge_rdds) but for now they can be thought of as simply RDDs of the form:
-`RDD[(VertexId, VD)]` and `RDD[Edge[ED]]`.
+`RDD[(VertexID, VD)]` and `RDD[Edge[ED]]`.
### Example Property Graph
@@ -168,7 +186,7 @@ code constructs a graph from a collection of RDDs:
// Assume the SparkContext has already been constructed
val sc: SparkContext
// Create an RDD for the vertices
-val users: RDD[(VertexID, (String, String))] =
+val users: RDD[(VertexId, (String, String))] =
sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
// Create an RDD for edges
@@ -183,7 +201,7 @@ val graph = Graph(users, relationships, defaultUser)
In the above example we make use of the [`Edge`][Edge] case class. Edges have a `srcId` and a
`dstId` corresponding to the source and destination vertex identifiers. In addition, the `Edge`
-class contains the `attr` member which contains the edge property.
+class has an `attr` member which stores the edge property.
[Edge]: api/graphx/index.html#org.apache.spark.graphx.Edge
@@ -199,7 +217,7 @@ graph.edges.filter(e => e.srcId > e.dstId).count
{% endhighlight %}
> Note that `graph.vertices` returns an `VertexRDD[(String, String)]` which extends
-> `RDD[(VertexId, (String, String))]` and so we use the scala `case` expression to deconstruct the
+> `RDD[(VertexID, (String, String))]` and so we use the scala `case` expression to deconstruct the
> tuple. On the other hand, `graph.edges` returns an `EdgeRDD` containing `Edge[String]` objects.
> We could have also used the case class type constructor as in the following:
> {% highlight scala %}
@@ -266,6 +284,75 @@ able to support different graph representations in the future. Each graph repre
provide implementations of the core operations and reuse many of the useful operations defined in
[`GraphOps`][GraphOps].
+### Summary List of Operators
+The following is a quick summary of the functionality defined in both [`Graph`][Graph] and
+[`GraphOps`][GraphOps] but presented as members of Graph for simplicity. Note that some function
+signatures have been simplified (e.g., default arguments and type constraints removed) and some more
+advanced functionality has been removed so please consult the API docs for the official list of
+operations.
+
+{% highlight scala %}
+/** Summary of the functionality in the property graph */
+class Graph[VD, ED] {
+ // Information about the Graph ===================================================================
+ val numEdges: Long
+ val numVertices: Long
+ val inDegrees: VertexRDD[Int]
+ val outDegrees: VertexRDD[Int]
+ val degrees: VertexRDD[Int]
+ // Views of the graph as collections =============================================================
+ val vertices: VertexRDD[VD]
+ val edges: EdgeRDD[ED]
+ val triplets: RDD[EdgeTriplet[VD, ED]]
+ // Functions for caching graphs ==================================================================
+ def persist(newLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]
+ def cache(): Graph[VD, ED]
+ def unpersistVertices(blocking: Boolean = true): Graph[VD, ED]
+ // Change the partitioning heuristic ============================================================
+ def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED]
+ // Transform vertex and edge attributes ==========================================================
+ def mapVertices[VD2](map: (VertexID, VD) => VD2): Graph[VD2, ED]
+ def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2]
+ def mapEdges[ED2](map: (PartitionID, Iterator[Edge[ED]]) => Iterator[ED2]): Graph[VD, ED2]
+ def mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2]
+ def mapTriplets[ED2](map: (PartitionID, Iterator[EdgeTriplet[VD, ED]]) => Iterator[ED2])
+ : Graph[VD, ED2]
+ // Modify the graph structure ====================================================================
+ def reverse: Graph[VD, ED]
+ def subgraph(
+ epred: EdgeTriplet[VD,ED] => Boolean = (x => true),
+ vpred: (VertexID, VD) => Boolean = ((v, d) => true))
+ : Graph[VD, ED]
+ def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
+ def groupEdges(merge: (ED, ED) => ED): Graph[VD, ED]
+ // Join RDDs with the graph ======================================================================
+ def joinVertices[U](table: RDD[(VertexID, U)])(mapFunc: (VertexID, VD, U) => VD): Graph[VD, ED]
+ def outerJoinVertices[U, VD2](other: RDD[(VertexID, U)])
+ (mapFunc: (VertexID, VD, Option[U]) => VD2)
+ : Graph[VD2, ED]
+ // Aggregate information about adjacent triplets =================================================
+ def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexID]]
+ def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[Array[(VertexID, VD)]]
+ def mapReduceTriplets[A: ClassTag](
+ mapFunc: EdgeTriplet[VD, ED] => Iterator[(VertexID, A)],
+ reduceFunc: (A, A) => A,
+ activeSetOpt: Option[(VertexRDD[_], EdgeDirection)] = None)
+ : VertexRDD[A]
+ // Iterative graph-parallel computation ==========================================================
+ def pregel[A](initialMsg: A, maxIterations: Int, activeDirection: EdgeDirection)(
+ vprog: (VertexID, VD, A) => VD,
+ sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexID,A)],
+ mergeMsg: (A, A) => A)
+ : Graph[VD, ED]
+ // Basic graph algorithms ========================================================================
+ def pageRank(tol: Double, resetProb: Double = 0.15): Graph[Double, Double]
+ def connectedComponents(): Graph[VertexID, ED]
+ def triangleCount(): Graph[Int, ED]
+ def stronglyConnectedComponents(numIter: Int): Graph[VertexID, ED]
+}
+{% endhighlight %}
+
+
## Property Operators
In direct analogy to the RDD `map` operator, the property
@@ -273,7 +360,7 @@ graph contains the following:
{% highlight scala %}
class Graph[VD, ED] {
- def mapVertices[VD2](map: (VertexID, VD) => VD2): Graph[VD2, 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]
}
@@ -295,7 +382,7 @@ val newGraph = Graph(newVertices, graph.edges)
val newGraph = graph.mapVertices((id, attr) => mapUdf(id, attr))
{% endhighlight %}
-[Graph.mapVertices]: api/graphx/index.html#org.apache.spark.graphx.Graph@mapVertices[VD2]((VertexID,VD)āVD2)(ClassTag[VD2]):Graph[VD2,ED]
+[Graph.mapVertices]: api/graphx/index.html#org.apache.spark.graphx.Graph@mapVertices[VD2]((VertexId,VD)āVD2)(ClassTag[VD2]):Graph[VD2,ED]
These operators are often used to initialize the graph for a particular computation or project away
unnecessary properties. For example, given a graph with the out-degrees as the vertex properties
@@ -321,7 +408,7 @@ add more in the future. The following is a list of the basic structural operato
class Graph[VD, ED] {
def reverse: Graph[VD, ED]
def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,
- vpred: (VertexID, VD) => Boolean): Graph[VD, ED]
+ vpred: (VertexId, VD) => Boolean): Graph[VD, ED]
def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
def groupEdges(merge: (ED, ED) => ED): Graph[VD,ED]
}
@@ -340,11 +427,11 @@ satisfy the edge predicate *and connect vertices that satisfy the vertex predica
operator can be used in number of situations to restrict the graph to the vertices and edges of
interest or eliminate broken links. For example in the following code we remove broken links:
-[Graph.subgraph]: api/graphx/index.html#org.apache.spark.graphx.Graph@subgraph((EdgeTriplet[VD,ED])āBoolean,(VertexID,VD)āBoolean):Graph[VD,ED]
+[Graph.subgraph]: api/graphx/index.html#org.apache.spark.graphx.Graph@subgraph((EdgeTriplet[VD,ED])āBoolean,(VertexId,VD)āBoolean):Graph[VD,ED]
{% highlight scala %}
// Create an RDD for the vertices
-val users: RDD[(VertexID, (String, String))] =
+val users: RDD[(VertexId, (String, String))] =
sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof")),
(4L, ("peter", "student"))))
@@ -407,9 +494,9 @@ using the *join* operators. Below we list the key join operators:
{% highlight scala %}
class Graph[VD, ED] {
- def joinVertices[U](table: RDD[(VertexID, U)])(map: (VertexID, VD, U) => VD)
+ def joinVertices[U](table: RDD[(VertexId, U)])(map: (VertexId, VD, U) => VD)
: Graph[VD, ED]
- def outerJoinVertices[U, VD2](table: RDD[(VertexID, U)])(map: (VertexID, VD, Option[U]) => VD2)
+ def outerJoinVertices[U, VD2](table: RDD[(VertexId, U)])(map: (VertexId, VD, Option[U]) => VD2)
: Graph[VD2, ED]
}
{% endhighlight %}
@@ -419,13 +506,13 @@ returns a new graph with the vertex properties obtained by applying the user def
to the result of the joined vertices. Vertices without a matching value in the RDD retain their
original value.
-[GraphOps.joinVertices]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@joinVertices[U](RDD[(VertexID,U)])((VertexID,VD,U)āVD)(ClassTag[U]):Graph[VD,ED]
+[GraphOps.joinVertices]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@joinVertices[U](RDD[(VertexId,U)])((VertexId,VD,U)āVD)(ClassTag[U]):Graph[VD,ED]
> Note that if the RDD contains more than one value for a given vertex only one will be used. It
> is therefore recommended that the input RDD be first made unique using the following which will
> also *pre-index* the resulting values to substantially accelerate the subsequent join.
> {% highlight scala %}
-val nonUniqueCosts: RDD[(VertexId, Double)]
+val nonUniqueCosts: RDD[(VertexID, Double)]
val uniqueCosts: VertexRDD[Double] =
graph.vertices.aggregateUsingIndex(nonUnique, (a,b) => a + b)
val joinedGraph = graph.joinVertices(uniqueCosts)(
@@ -438,7 +525,7 @@ property type. Because not all vertices may have a matching value in the input
function takes an `Option` type. For example, we can setup a graph for PageRank by initializing
vertex properties with their `outDegree`.
-[Graph.outerJoinVertices]: api/graphx/index.html#org.apache.spark.graphx.Graph@outerJoinVertices[U,VD2](RDD[(VertexID,U)])((VertexID,VD,Option[U])āVD2)(ClassTag[U],ClassTag[VD2]):Graph[VD2,ED]
+[Graph.outerJoinVertices]: api/graphx/index.html#org.apache.spark.graphx.Graph@outerJoinVertices[U,VD2](RDD[(VertexId,U)])((VertexId,VD,Option[U])āVD2)(ClassTag[U],ClassTag[VD2]):Graph[VD2,ED]
{% highlight scala %}
@@ -457,7 +544,7 @@ val degreeGraph = graph.outerJoinVertices(outDegrees) { (id, oldAttr, outDegOpt)
> provide type annotation for the user defined function:
> {% highlight scala %}
val joinedGraph = graph.joinVertices(uniqueCosts,
- (id: VertexId, oldCost: Double, extraCost: Double) => oldCost + extraCost)
+ (id: VertexID, oldCost: Double, extraCost: Double) => oldCost + extraCost)
{% endhighlight %}
@@ -472,7 +559,7 @@ PageRank Value, shortest path to the source, and smallest reachable vertex id).
### Map Reduce Triplets (mapReduceTriplets)
-[Graph.mapReduceTriplets]: api/graphx/index.html#org.apache.spark.graphx.Graph@mapReduceTriplets[A](mapFunc:org.apache.spark.graphx.EdgeTriplet[VD,ED]=>Iterator[(org.apache.spark.graphx.VertexID,A)],reduceFunc:(A,A)=>A,activeSetOpt:Option[(org.apache.spark.graphx.VertexRDD[_],org.apache.spark.graphx.EdgeDirection)])(implicitevidence$10:scala.reflect.ClassTag[A]):org.apache.spark.graphx.VertexRDD[A]
+[Graph.mapReduceTriplets]: api/graphx/index.html#org.apache.spark.graphx.Graph@mapReduceTriplets[A](mapFunc:org.apache.spark.graphx.EdgeTriplet[VD,ED]=>Iterator[(org.apache.spark.graphx.VertexId,A)],reduceFunc:(A,A)=>A,activeSetOpt:Option[(org.apache.spark.graphx.VertexRDD[_],org.apache.spark.graphx.EdgeDirection)])(implicitevidence$10:scala.reflect.ClassTag[A]):org.apache.spark.graphx.VertexRDD[A]
The core (heavily optimized) aggregation primitive in GraphX is the
[`mapReduceTriplets`][Graph.mapReduceTriplets] operator:
@@ -480,7 +567,7 @@ The core (heavily optimized) aggregation primitive in GraphX is the
{% highlight scala %}
class Graph[VD, ED] {
def mapReduceTriplets[A](
- map: EdgeTriplet[VD, ED] => Iterator[(VertexID, A)],
+ map: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
reduce: (A, A) => A)
: VertexRDD[A]
}
@@ -495,26 +582,26 @@ containing the aggregate message (of type `A`) destined to each vertex. Vertice
receive a message are not included in the returned `VertexRDD`.
-In the following example we use the `mapReduceTriplets` operator to compute the average age of the @@ -547,8 +634,8 @@ val avgAgeOfOlderFollowers: VertexRDD[Double] = avgAgeOfOlderFollowers.collect.foreach(println(_)) {% endhighlight %} -> Note that the `mapReduceTriplets` operation performs optimally when the messages (and their sums) -> are constant sized (e.g., floats and addition instead of lists and concatenation). More +> Note that the `mapReduceTriplets` operation performs optimally when the messages (and the sums of +> messages) are constant sized (e.g., floats and addition instead of lists and concatenation). More > precisely, the result of `mapReduceTriplets` should ideally be sub-linear in the degree of each > vertex. @@ -562,13 +649,13 @@ compute the max in, out, and total degrees: {% highlight scala %} // Define a reduce operation to compute the highest degree vertex -def max(a: (VertexID, Int), b: (VertexID, Int)): (VertexID, Int) = { +def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = { if (a._2 > b._2) a else b } // Compute the max degrees -val maxInDegree: (VertexID, Int) = graph.inDegrees.reduce(max) -val maxOutDegree: (VertexID, Int) = graph.outDegrees.reduce(max) -val maxDegrees: (VertexID, Int) = graph.degrees.reduce(max) +val maxInDegree: (VertexId, Int) = graph.inDegrees.reduce(max) +val maxOutDegree: (VertexId, Int) = graph.outDegrees.reduce(max) +val maxDegrees: (VertexId, Int) = graph.degrees.reduce(max) {% endhighlight %} ### Collecting Neighbors @@ -578,14 +665,14 @@ attributes at each vertex. This can be easily accomplished using the [`collectNeighborIds`][GraphOps.collectNeighborIds] and the [`collectNeighbors`][GraphOps.collectNeighbors] operators. -[GraphOps.collectNeighborIds]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@collectNeighborIds(EdgeDirection):VertexRDD[Array[VertexID]] -[GraphOps.collectNeighbors]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@collectNeighbors(EdgeDirection):VertexRDD[Array[(VertexID,VD)]] +[GraphOps.collectNeighborIds]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@collectNeighborIds(EdgeDirection):VertexRDD[Array[VertexId]] +[GraphOps.collectNeighbors]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@collectNeighbors(EdgeDirection):VertexRDD[Array[(VertexId,VD)]] {% highlight scala %} class GraphOps[VD, ED] { - def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexID]] - def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[ Array[(VertexID, VD)] ] + def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexId]] + def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[ Array[(VertexId, VD)] ] } {% endhighlight %} @@ -593,11 +680,20 @@ class GraphOps[VD, ED] { > substantial communication. If possible try expressing the same computation using the > `mapReduceTriplets` operator directly. +## Caching and Uncaching + +In Spark, RDDs are not persisted in memory by default. To avoid recomputation, they must be explicitly cached when using them multiple times (see the [Spark Programming Guide][RDD Persistence]). Graphs in GraphX behave the same way. **When using a graph multiple times, make sure to call [`Graph.cache()`][Graph.cache] on it first.** + +[RDD Persistence]: scala-programming-guide.html#rdd-persistence +[Graph.cache]: api/graphx/index.html#org.apache.spark.graphx.Graph@cache():Graph[VD,ED] + +In iterative computations, *uncaching* may also be necessary for best performance. By default, cached RDDs and graphs will remain in memory until memory pressure forces them to be evicted in LRU order. For iterative computation, intermediate results from previous iterations will fill up the cache. Though they will eventually be evicted, the unnecessary data stored in memory will slow down garbage collection. It would be more efficient to uncache intermediate results as soon as they are no longer necessary. This involves materializing (caching and forcing) a graph or RDD every iteration, uncaching all other datasets, and only using the materialized dataset in future iterations. However, because graphs are composed of multiple RDDs, it can be difficult to unpersist them correctly. **For iterative computation we recommend using the Pregel API, which correctly unpersists intermediate results.** + # Pregel API Graphs are inherently recursive data-structures as properties of vertices depend on properties of -their neighbors which intern depend on properties of *their* neighbors. As a +their neighbors which in turn depend on properties of *their* neighbors. As a consequence many important graph algorithms iteratively recompute the properties of each vertex until a fixed-point condition is reached. A range of graph-parallel abstractions have been proposed to express these iterative algorithms. GraphX exposes a Pregel-like operator which is a fusion of @@ -620,7 +716,7 @@ messages remaining. The following is the type signature of the [Pregel operator][GraphOps.pregel] as well as a *sketch* of its implementation (note calls to graph.cache have been removed): -[GraphOps.pregel]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@pregel[A](A,Int,EdgeDirection)((VertexID,VD,A)āVD,(EdgeTriplet[VD,ED])āIterator[(VertexID,A)],(A,A)āA)(ClassTag[A]):Graph[VD,ED] +[GraphOps.pregel]: api/graphx/index.html#org.apache.spark.graphx.GraphOps@pregel[A](A,Int,EdgeDirection)((VertexId,VD,A)āVD,(EdgeTriplet[VD,ED])āIterator[(VertexId,A)],(A,A)āA)(ClassTag[A]):Graph[VD,ED] {% highlight scala %} class GraphOps[VD, ED] { @@ -628,8 +724,8 @@ class GraphOps[VD, ED] { (initialMsg: A, maxIter: Int = Int.MaxValue, activeDir: EdgeDirection = EdgeDirection.Out) - (vprog: (VertexID, VD, A) => VD, - sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexID, A)], + (vprog: (VertexId, VD, A) => VD, + sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)], mergeMsg: (A, A) => A) : Graph[VD, ED] = { // Receive the initial message at each vertex @@ -674,7 +770,7 @@ import org.apache.spark.graphx.util.GraphGenerators // A graph with edge attributes containing distances val graph: Graph[Int, Double] = GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble) -val sourceId: VertexID = 42 // The ultimate source +val sourceId: VertexId = 42 // The ultimate source // Initialize the graph such that all vertices except the root have distance infinity. val initialGraph = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity) val sssp = initialGraph.pregel(Double.PositiveInfinity)( @@ -721,7 +817,7 @@ It creates a `Graph` from the specified edges, automatically creating any vertic {% highlight scala %} object Graph { def apply[VD, ED]( - vertices: RDD[(VertexID, VD)], + vertices: RDD[(VertexId, VD)], edges: RDD[Edge[ED]], defaultVertexAttr: VD = null) : Graph[VD, ED] @@ -731,7 +827,7 @@ object Graph { defaultValue: VD): Graph[VD, ED] def fromEdgeTuples[VD]( - rawEdges: RDD[(VertexID, VertexID)], + rawEdges: RDD[(VertexId, VertexId)], defaultValue: VD, uniqueEdges: Option[PartitionStrategy] = None): Graph[VD, Int] @@ -747,8 +843,8 @@ object Graph { [PartitionStrategy]: api/graphx/index.html#org.apache.spark.graphx.PartitionStrategy$ [GraphLoader.edgeListFile]: api/graphx/index.html#org.apache.spark.graphx.GraphLoader$@edgeListFile(SparkContext,String,Boolean,Int):Graph[Int,Int] -[Graph.apply]: api/graphx/index.html#org.apache.spark.graphx.Graph$@apply[VD,ED](RDD[(VertexID,VD)],RDD[Edge[ED]],VD)(ClassTag[VD],ClassTag[ED]):Graph[VD,ED] -[Graph.fromEdgeTuples]: api/graphx/index.html#org.apache.spark.graphx.Graph$@fromEdgeTuples[VD](RDD[(VertexID,VertexID)],VD,Option[PartitionStrategy])(ClassTag[VD]):Graph[VD,Int] +[Graph.apply]: api/graphx/index.html#org.apache.spark.graphx.Graph$@apply[VD,ED](RDD[(VertexId,VD)],RDD[Edge[ED]],VD)(ClassTag[VD],ClassTag[ED]):Graph[VD,ED] +[Graph.fromEdgeTuples]: api/graphx/index.html#org.apache.spark.graphx.Graph$@fromEdgeTuples[VD](RDD[(VertexId,VertexId)],VD,Option[PartitionStrategy])(ClassTag[VD]):Graph[VD,Int] [Graph.fromEdges]: api/graphx/index.html#org.apache.spark.graphx.Graph$@fromEdges[VD,ED](RDD[Edge[ED]],VD)(ClassTag[VD],ClassTag[ED]):Graph[VD,ED] # Vertex and Edge RDDs @@ -761,47 +857,46 @@ respectively. In this section we review some of the additional useful functiona ## VertexRDDs -The `VertexRDD[A]` extends the more traditional `RDD[(VertexId, A)]` but adds the additional -constraint that each `VertexId` occurs only *once*. Moreover, `VertexRDD[A]` represents a *set* of -vertices each with an attribute of type `A`. Internally, this is achieved by storing the vertex -attributes in a reusable hash-map data-structure. As a consequence if two `VertexRDD`s are derived -from the same base `VertexRDD` (e.g., by `filter` or `mapValues`) they can be joined in constant -time without hash evaluations. To leverage this indexed data-structure, the `VertexRDD` exposes the -following additional functionality: +The `VertexRDD[A]` extends `RDD[(VertexID, A)]` and adds the additional constraint that each +`VertexID` occurs only *once*. Moreover, `VertexRDD[A]` represents a *set* of vertices each with an +attribute of type `A`. Internally, this is achieved by storing the vertex attributes in a reusable +hash-map data-structure. As a consequence if two `VertexRDD`s are derived from the same base +`VertexRDD` (e.g., by `filter` or `mapValues`) they can be joined in constant time without hash +evaluations. To leverage this indexed data-structure, the `VertexRDD` exposes the following +additional functionality: {% highlight scala %} -class VertexRDD[VD] { +class VertexRDD[VD] extends RDD[(VertexID, VD)] { // Filter the vertex set but preserves the internal index - def filter(pred: Tuple2[VertexID, VD] => Boolean): VertexRDD[VD] + def filter(pred: Tuple2[VertexId, VD] => Boolean): VertexRDD[VD] // Transform the values without changing the ids (preserves the internal index) def mapValues[VD2](map: VD => VD2): VertexRDD[VD2] - def mapValues[VD2](map: (VertexID, VD) => VD2): VertexRDD[VD2] + def mapValues[VD2](map: (VertexId, VD) => VD2): VertexRDD[VD2] // Remove vertices from this set that appear in the other set def diff(other: VertexRDD[VD]): VertexRDD[VD] // Join operators that take advantage of the internal indexing to accelerate joins (substantially) - def leftJoin[VD2, VD3](other: RDD[(VertexID, VD2)])(f: (VertexID, VD, Option[VD2]) => VD3): VertexRDD[VD3] - def innerJoin[U, VD2](other: RDD[(VertexID, U)])(f: (VertexID, VD, U) => VD2): VertexRDD[VD2] + def leftJoin[VD2, VD3](other: RDD[(VertexId, VD2)])(f: (VertexId, VD, Option[VD2]) => VD3): VertexRDD[VD3] + def innerJoin[U, VD2](other: RDD[(VertexId, U)])(f: (VertexId, VD, U) => VD2): VertexRDD[VD2] // Use the index on this RDD to accelerate a `reduceByKey` operation on the input RDD. - def aggregateUsingIndex[VD2](other: RDD[(VertexID, VD2)], reduceFunc: (VD2, VD2) => VD2): VertexRDD[VD2] + def aggregateUsingIndex[VD2](other: RDD[(VertexId, VD2)], reduceFunc: (VD2, VD2) => VD2): VertexRDD[VD2] } {% endhighlight %} Notice, for example, how the `filter` operator returns an `VertexRDD`. Filter is actually implemented using a `BitSet` thereby reusing the index and preserving the ability to do fast joins with other `VertexRDD`s. Likewise, the `mapValues` operators do not allow the `map` function to -change the `VertexId` thereby enabling the same `HashMap` data-structures to be reused. Both the +change the `VertexID` thereby enabling the same `HashMap` data-structures to be reused. Both the `leftJoin` and `innerJoin` are able to identify when joining two `VertexRDD`s derived from the same `HashMap` and implement the join by linear scan rather than costly point lookups. -The `aggregateUsingIndex` operator can be slightly confusing but is also useful for efficient -construction of a new `VertexRDD` from an `RDD[(VertexId, A)]`. Conceptually, if I have constructed -a `VertexRDD[B]` over a set of vertices, *which is a super-set* of the vertices in some -`RDD[(VertexId, A)]` then I can reuse the index to both aggregate and then subsequently index the -RDD. For example: +The `aggregateUsingIndex` operator is useful for efficient construction of a new `VertexRDD` from an +`RDD[(VertexID, A)]`. Conceptually, if I have constructed a `VertexRDD[B]` over a set of vertices, +*which is a super-set* of the vertices in some `RDD[(VertexID, A)]` then I can reuse the index to +both aggregate and then subsequently index the `RDD[(VertexID, A)]`. For example: {% highlight scala %} val setA: VertexRDD[Int] = VertexRDD(sc.parallelize(0L until 100L).map(id => (id, 1))) -val rddB: RDD[(VertexID, Double)] = sc.parallelize(0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0))) +val rddB: RDD[(VertexId, Double)] = sc.parallelize(0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0))) // There should be 200 entries in rddB rddB.count val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ + _) @@ -813,10 +908,10 @@ val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b) => a + b) ## EdgeRDDs -The `EdgeRDD[ED]`, which extends `RDD[Edge[ED]]` is considerably simpler than the `VertexRDD`. -GraphX organizes the edges in blocks partitioned using one of the various partitioning strategies -defined in [`PartitionStrategy`][PartitionStrategy]. Within each partition, edge attributes and -adjacency structure, are stored separately enabling maximum reuse when changing attribute values. +The `EdgeRDD[ED]`, which extends `RDD[Edge[ED]]` organizes the edges in blocks partitioned using one +of the various partitioning strategies defined in [`PartitionStrategy`][PartitionStrategy]. Within +each partition, edge attributes and adjacency structure, are stored separately enabling maximum +reuse when changing attribute values. [PartitionStrategy]: api/graphx/index.html#org.apache.spark.graphx.PartitionStrategy @@ -827,11 +922,11 @@ def mapValues[ED2](f: Edge[ED] => ED2): EdgeRDD[ED2] // Revere the edges reusing both attributes and structure def reverse: EdgeRDD[ED] // Join two `EdgeRDD`s partitioned using the same partitioning strategy. -def innerJoin[ED2, ED3](other: EdgeRDD[ED2])(f: (VertexID, VertexID, ED, ED2) => ED3): EdgeRDD[ED3] +def innerJoin[ED2, ED3](other: EdgeRDD[ED2])(f: (VertexId, VertexId, ED, ED2) => ED3): EdgeRDD[ED3] {% endhighlight %} In most applications we have found that operations on the `EdgeRDD` are accomplished through the -graph or rely on operations defined in the base `RDD` class. +graph operators or rely on operations defined in the base `RDD` class. # Optimized Representation @@ -853,7 +948,9 @@ reduce both the communication and storage overhead. Logically, this corresponds to machines and allowing vertices to span multiple machines. The exact method of assigning edges depends on the [`PartitionStrategy`][PartitionStrategy] and there are several tradeoffs to the various heuristics. Users can choose between different strategies by repartitioning the graph with -the [`Graph.partitionBy`][Graph.partitionBy] operator. +the [`Graph.partitionBy`][Graph.partitionBy] operator. The default partitioning strategy is to use +the initial partitioning of the edges as provided on graph construction. However, users can easily +switch to 2D-partitioning or other heuristics included in GraphX. [Graph.partitionBy]: api/graphx/index.html#org.apache.spark.graphx.Graph$@partitionBy(partitionStrategy:org.apache.spark.graphx.PartitionStrategy):org.apache.spark.graphx.Graph[VD,ED] @@ -867,16 +964,15 @@ the [`Graph.partitionBy`][Graph.partitionBy] operator. Once the edges have be partitioned the key challenge to efficient graph-parallel computation is efficiently joining vertex attributes with the edges. Because real-world graphs typically have more -edges than vertices, we move vertex attributes to the edges. - - - - +edges than vertices, we move vertex attributes to the edges. Because not all partitions will +contain edges adjacent to all vertices we internally maintain a routing table which identifies where +to broadcast vertices when implementing the join required for operations like `triplets` and +`mapReduceTriplets`. # Graph Algorithms -GraphX includes a set of graph algorithms in to simplify analytics. The algorithms are contained in the `org.apache.spark.graphx.lib` package and can be accessed directly as methods on `Graph` via [`GraphOps`][GraphOps]. This section describes the algorithms and how they are used. +GraphX includes a set of graph algorithms to simplify analytics tasks. The algorithms are contained in the `org.apache.spark.graphx.lib` package and can be accessed directly as methods on `Graph` via [`GraphOps`][GraphOps]. This section describes the algorithms and how they are used. ## PageRank @@ -953,13 +1049,6 @@ val triCountByUsername = users.join(triCounts).map { case (id, (username, tc)) = println(triCountByUsername.collect().mkString("\n")) {% endhighlight %} --Note that
+ +mapReduceTripletstakes an additional optionalactiveSet-(see API docs) which restricts the map phase to edges adjacent to the vertices in the provided -VertexRDD: -Note that
+ {% highlight scala %} activeSetOpt: Option[(VertexRDD[_], EdgeDirection)] = None {% endhighlight %} -mapReduceTripletstakes an additional optionalactiveSet+(not shown above see API docs for details) which restricts the map phase to edges adjacent to the +vertices in the providedVertexRDD:-The EdgeDirection specifies which edges adjacent to the vertex set are included in the map phase. If -the direction is
+ +In,mapFuncwill only be run only on edges with -destination in the active set. If the direction isOut,mapFuncwill only -be run only on edges originating from vertices in the active set. If the direction is -Either,mapFuncwill be run only on edges with either vertex in the -active set. If the direction isBoth,mapFuncwill be run only on edges -with both vertices in the active set. The active set must be derived from the set of vertices in -the graph. Restricting computation to triplets adjacent to a subset of the vertices is often -necessary in incremental iterative computation and is a key part of the GraphX implementation of -Pregel. -The EdgeDirection specifies which edges adjacent to the vertex set are included in the map +phase. If the direction is
+In, then the user definedmapfunction will +only be run only on edges with the destination vertex in the active set. If the direction is +Out, then themapfunction will only be run only on edges originating from +vertices in the active set. If the direction isEither, then themap+function will be run only on edges with either vertex in the active set. If the direction is +Both, then themapfunction will be run only on edges with both vertices +in the active set. The active set must be derived from the set of vertices in the graph. +Restricting computation to triplets adjacent to a subset of the vertices is often necessary in +incremental iterative computation and is a key part of the GraphX implementation of Pregel.
-
-
-
' +
+ '' +
+ title +
+ '' +
+ '
'
+ );
+}
+
+function viewSolution() {
+ var counter = 0
+ $("div.solution").each(function() {
+ var id = "solution_" + counter
+ makeCollapsable(this, "", id,
+ '' +
+ '' + "View Solution");
+ counter++;
+ });
+}
+
+
+$(function() {
+ codeTabs();
+ viewSolution();
+});
diff --git a/docs/mllib-guide.md b/docs/mllib-guide.md
index 1a5c640d10df4..a22a22184b5c6 100644
--- a/docs/mllib-guide.md
+++ b/docs/mllib-guide.md
@@ -21,7 +21,8 @@ depends on native Fortran routines. You may need to install the
if it is not already present on your nodes. MLlib will throw a linking error if it cannot
detect these libraries automatically.
-To use MLlib in Python, you will also need [NumPy](http://www.numpy.org) version 1.7 or newer.
+To use MLlib in Python, you will need [NumPy](http://www.numpy.org) version 1.7 or newer
+and Python 2.7.
# Binary Classification
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 0d5eb7065e9f0..242318550c127 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -48,11 +48,22 @@ Each instance can report to zero or more _sinks_. Sinks are contained in the
* `ConsoleSink`: Logs metrics information to the console.
* `CSVSink`: Exports metrics data to CSV files at regular intervals.
-* `GangliaSink`: Sends metrics to a Ganglia node or multicast group.
* `JmxSink`: Registers metrics for viewing in a JXM console.
* `MetricsServlet`: Adds a servlet within the existing Spark UI to serve metrics data as JSON data.
* `GraphiteSink`: Sends metrics to a Graphite node.
+Spark also supports a Ganglia sink which is not included in the default build due to
+licensing restrictions:
+
+* `GangliaSink`: Sends metrics to a Ganglia node or multicast group.
+
+To install the `GangliaSink` you'll need to perform a custom build of Spark. _**Note that
+by embedding this library you will include [LGPL](http://www.gnu.org/copyleft/lesser.html)-licensed
+code in your Spark package**_. For sbt users, set the
+`SPARK_GANGLIA_LGPL` environment variable before building. For Maven users, enable
+the `-Pspark-ganglia-lgpl` profile. In addition to modifying the cluster's Spark build
+user applications will need to link to the `spark-ganglia-lgpl` artifact.
+
The syntax of the metrics configuration file is defined in an example configuration file,
`$SPARK_HOME/conf/metrics.conf.template`.
diff --git a/docs/python-programming-guide.md b/docs/python-programming-guide.md
index b07899c2e176d..7c5283fb0b6fb 100644
--- a/docs/python-programming-guide.md
+++ b/docs/python-programming-guide.md
@@ -52,7 +52,7 @@ In addition, PySpark fully supports interactive use---simply run `./bin/pyspark`
# Installing and Configuring PySpark
-PySpark requires Python 2.7 or higher.
+PySpark requires Python 2.6 or higher.
PySpark applications are executed using a standard CPython interpreter in order to support Python modules that use C extensions.
We have not tested PySpark with Python 3 or with alternative Python interpreters, such as [PyPy](http://pypy.org/) or [Jython](http://www.jython.org/).
@@ -152,7 +152,7 @@ Many of the methods also contain [doctests](http://docs.python.org/2/library/doc
# Libraries
[MLlib](mllib-guide.html) is also available in PySpark. To use it, you'll need
-[NumPy](http://www.numpy.org) version 1.7 or newer. The [MLlib guide](mllib-guide.html) contains
+[NumPy](http://www.numpy.org) version 1.7 or newer, and Python 2.7. The [MLlib guide](mllib-guide.html) contains
some example applications.
# Where to Go from Here
diff --git a/docs/quick-start.md b/docs/quick-start.md
index 153081bdaa286..13df6beea16e8 100644
--- a/docs/quick-start.md
+++ b/docs/quick-start.md
@@ -115,7 +115,7 @@ object SimpleApp {
def main(args: Array[String]) {
val logFile = "$YOUR_SPARK_HOME/README.md" // Should be some file on your system
val sc = new SparkContext("local", "Simple App", "YOUR_SPARK_HOME",
- List("target/scala-{{site.SCALA_VERSION}}/simple-project_{{site.SCALA_VERSION}}-1.0.jar"))
+ List("target/scala-{{site.SCALA_BINARY_VERSION}}/simple-project_{{site.SCALA_BINARY_VERSION}}-1.0.jar"))
val logData = sc.textFile(logFile, 2).cache()
val numAs = logData.filter(line => line.contains("a")).count()
val numBs = logData.filter(line => line.contains("b")).count()
@@ -214,7 +214,7 @@ To build the program, we also write a Maven `pom.xml` file that lists Spark as a
+ 
+
+ 
+
+
+
+If you have already [downloaded](index.html#downloading) and [built](index.html#building) Spark,
+you can run this example as follows. You will first need to run Netcat
+(a small utility found in most Unix-like systems) as a data server by using
+
+{% highlight bash %}
+$ nc -lk 9999
+{% endhighlight %}
+
+Then, in a different terminal, you can start the example by using
+
+
+
+First, we create a
+[StreamingContext](api/streaming/index.html#org.apache.spark.streaming.StreamingContext) object,
+which is the main entry point for all streaming
+functionality. Besides Spark's configuration, we specify that any DStream will be processed
+in 1 second batches.
-This guide shows some how to start programming with DStreams.
+{% highlight scala %}
+// Create a StreamingContext with a SparkConf configuration
+val ssc = new StreamingContext(sparkConf, Seconds(1))
+{% endhighlight %}
-# Linking with Spark Streaming
+Using this context, we then create a new DStream
+by specifying the IP address and port of the data server.
-Add the following SBT or Maven dependency to your project to use Spark Streaming:
+{% highlight scala %}
+// Create a DStream that will connect to serverIP:serverPort
+val lines = ssc.socketTextStream(serverIP, serverPort)
+{% endhighlight %}
- groupId = org.apache.spark
- artifactId = spark-streaming_{{site.SCALA_VERSION}}
- version = {{site.SPARK_VERSION}}
+This `lines` DStream represents the stream of data that will be received from the data
+server. Each record in this DStream is a line of text. Next, we want to split the lines by
+space into words.
-For ingesting data from sources like Kafka and Flume, add the corresponding artifact `spark-streaming-xyz_{{site.SCALA_VERSION}}` to the dependencies. For example, `spark-streaming-kafka_{{site.SCALA_VERSION}}` for Kafka, `spark-streaming-flume_{{site.SCALA_VERSION}}`, etc. Please refer to the [Apache repository](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20AND%20v%3A%22{{site.SPARK_VERSION}}%22) for the full list of supported sources / artifacts.
+{% highlight scala %}
+// Split each line into words
+val words = lines.flatMap(_.split(" "))
+{% endhighlight %}
-# Initializing Spark Streaming
-The first thing a Spark Streaming program must do is create a `StreamingContext` object, which tells Spark how to access a cluster. A `StreamingContext` can be created by using
+`flatMap` is a one-to-many DStream operation that creates a new DStream by
+generating multiple new records from each record int the source DStream. In this case,
+each line will be split into multiple words and the stream of words is represented as the
+`words` DStream. Next, we want to count these words.
{% highlight scala %}
-new StreamingContext(master, appName, batchDuration, [sparkHome], [jars])
+// Count each word in each batch
+val pairs = words.map(word => (word, 1))
+val wordCounts = pairs.reduceByKey(_ + _)
+
+// Print a few of the counts to the console
+wordCount.print()
{% endhighlight %}
-The `master` parameter is a standard [Spark cluster URL](scala-programming-guide.html#master-urls) and can be "local" for local testing. The `appName` is a name of your program, which will be shown on your cluster's web UI. The `batchDuration` is the size of the batches (as explained earlier). This must be set carefully such that the cluster can keep up with the processing of the data streams. Start with something conservative like 5 seconds. See the [Performance Tuning](#setting-the-right-batch-size) section for a detailed discussion. Finally, `sparkHome` and `jars` are optional parameters, which need to be set when running on a cluster to specify the location of your code, as described in the [Spark programming guide](scala-programming-guide.html#deploying-code-on-a-cluster).
+The `words` DStream is further mapped (one-to-one transformation) to a DStream of `(word,
+1)` pairs, which is then reduced to get the frequency of words in each batch of data.
+Finally, `wordCounts.print()` will print a few of the counts generated every second.
+
+Note that when these lines are executed, Spark Streaming only sets up the computation it
+will perform when it is started, and no real processing has started yet. To start the processing
+after all the transformations have been setup, we finally call
{% highlight scala %}
-new SparkConf(conf, batchDuration)
+ssc.start() // Start the computation
+ssc.awaitTermination() // Wait for the computation to terminate
+{% endhighlight %}
+
+The complete code can be found in the Spark Streaming example
+[NetworkWordCount]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/scala/org/apache/spark/streaming/examples/NetworkWordCount.scala).
+
+ +
++ +
+
+First, we create a
+[JavaStreamingContext](api/streaming/index.html#org.apache.spark.streaming.api.java.JavaStreamingContext) object,
+which is the main entry point for all streaming
+functionality. Besides Spark's configuration, we specify that any DStream would be processed
+in 1 second batches.
+
+{% highlight java %}
+// Create a StreamingContext with a SparkConf configuration
+JavaStreamingContext jssc = StreamingContext(sparkConf, new Duration(1000))
+{% endhighlight %}
+
+Using this context, we then create a new DStream
+by specifying the IP address and port of the data server.
+
+{% highlight java %}
+// Create a DStream that will connect to serverIP:serverPort
+JavaDStream lines = jssc.socketTextStream(serverIP, serverPort);
+{% endhighlight %}
+
+This `lines` DStream represents the stream of data that will be received from the data
+server. Each record in this stream is a line of text. Then, we want to split the the lines by
+space into words.
+
+{% highlight java %}
+// Split each line into words
+JavaDStream words = lines.flatMap(
+ new FlatMapFunction() {
+ @Override public Iterable call(String x) {
+ return Lists.newArrayList(x.split(" "));
+ }
+ });
+{% endhighlight %}
+
+`flatMap` is a DStream operation that creates a new DStream by
+generating multiple new records from each record in the source DStream. In this case,
+each line will be split into multiple words and the stream of words is represented as the
+`words` DStream. Note that we defined the transformation using a
+[FlatMapFunction](api/core/index.html#org.apache.spark.api.java.function.FlatMapFunction) object.
+As we will discover along the way, there are a number of such convenience classes in the Java API
+that help define DStream transformations.
+
+Next, we want to count these words.
+
+{% highlight java %}
+// Count each word in each batch
+JavaPairDStream pairs = words.map(
+ new PairFunction() {
+ @Override public Tuple2 call(String s) throws Exception {
+ return new Tuple2(s, 1);
+ }
+ });
+JavaPairDStream wordCounts = pairs.reduceByKey(
+ new Function2() {
+ @Override public Integer call(Integer i1, Integer i2) throws Exception {
+ return i1 + i2;
+ }
+ });
+wordCount.print(); // Print a few of the counts to the console
{% endhighlight %}
-where `conf` is a [SparkConf](api/core/index.html#org.apache.spark.SparkConf)
-object used for more advanced configuration. In both cases, a [SparkContext](api/core/index.html#org.apache.spark.SparkContext) is created as well which can be accessed with `streamingContext.sparkContext`.
+The `words` DStream is further mapped (one-to-one transformation) to a DStream of `(word,
+1)` pairs, using a [PairFunction](api/core/index.html#org.apache.spark.api.java.function.PairFunction)
+object. Then, it is reduced to get the frequency of words in each batch of data,
+using a [Function2](api/core/index.html#org.apache.spark.api.java.function.Function2) object.
+Finally, `wordCounts.print()` will print a few of the counts generated every second.
+
+Note that when these lines are executed, Spark Streaming only sets up the computation it
+will perform when it is started, and no real processing has started yet. To start the processing
+after all the transformations have been setup, we finally call
-# Attaching Input Sources
-The StreamingContext is used to creating input streams from data sources:
+{% highlight java %}
+jssc.start(); // Start the computation
+jssc.awaitTermination(); // Wait for the computation to terminate
+{% endhighlight %}
+
+The complete code can be found in the Spark Streaming example
+[JavaNetworkWordCount]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/java/org/apache/spark/streaming/examples/JavaNetworkWordCount.java).
+
+ +
++ +
+
+
+
+Then, any lines typed in the terminal running the netcat server will be counted and printed on
+screen every second. It will look something like this.
+
+
+{% highlight bash %}
+$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999
+{% endhighlight %}
+
+
+{% highlight bash %}
+$ ./bin/run-example org.apache.spark.streaming.examples.JavaNetworkWordCount local[2] localhost 9999
+{% endhighlight %}
+
+| +{% highlight bash %} +# TERMINAL 1: +# Running Netcat + +$ nc -lk 9999 + +hello world + + + +... +{% endhighlight %} + | ++ | +{% highlight bash %} +# TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount + +$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999 +... +------------------------------------------- +Time: 1357008430000 ms +------------------------------------------- +(hello,1) +(world,1) +... +{% endhighlight %} + | +
| Source | Artifact |
|---|---|
| Kafka | spark-streaming-kafka_{{site.SCALA_BINARY_VERSION}} |
| Flume | spark-streaming-flume_{{site.SCALA_BINARY_VERSION}} |
| spark-streaming-twitter_{{site.SCALA_BINARY_VERSION}} | |
| ZeroMQ | spark-streaming-zeromq_{{site.SCALA_BINARY_VERSION}} |
| MQTT | spark-streaming-mqtt_{{site.SCALA_BINARY_VERSION}} |
+
+
+The `master` parameter is a standard [Spark cluster URL](scala-programming-guide.html#master-urls)
+and can be "local" for local testing. The `appName` is a name of your program,
+which will be shown on your cluster's web UI. The `batchInterval` is the size of the batches,
+as explained earlier. Finally, the last two parameters are needed to deploy your code to a cluster
+ if running in distributed mode, as described in the
+ [Spark programming guide](scala-programming-guide.html#deploying-code-on-a-cluster).
+ Additionally, the underlying SparkContext can be accessed as
+`streamingContext.sparkContext`.
+
+The batch interval must be set based on the latency requirements of your application
+and available cluster resources. See the [Performance Tuning](#setting-the-right-batch-size)
+section for more details.
+
+## DStreams
+*Discretized Stream* or *DStream* is the basic abstraction provided by Spark Streaming.
+It represents a continuous stream of data, either the input data stream received from source,
+or the processed data stream generated by transforming the input stream. Internally,
+it is represented by a continuous sequence of RDDs, which is Spark's abstraction of an immutable,
+distributed dataset. Each RDD in a DStream contains data from a certain interval,
+as shown in the following figure.
+
+
+
+To initialize a Spark Streaming program in Scala, a
+[`StreamingContext`](api/streaming/index.html#org.apache.spark.streaming.StreamingContext)
+object has to be created, which is the main entry point of all Spark Streaming functionality.
+A `StreamingContext` object can be created by using
{% highlight scala %}
-// Assuming ssc is the StreamingContext
-ssc.textFileStream(directory) // Creates a stream that monitors and processes new files in a HDFS directory
-ssc.socketStream(hostname, port) // Creates a stream that uses a TCP socket to read data from hostname:port
+new StreamingContext(master, appName, batchDuration, [sparkHome], [jars])
{% endhighlight %}
+
+
-The core Spark Streaming API provides input streams for files, sockets, and Akka actors. Additional functionality for Kafka, Flume, ZeroMQ, Twitter, etc. can be imported by adding the right dependencies as explained in the [linking](#linking-with-spark-streaming) section.
+To initialize a Spark Streaming program in Java, a
+[`JavaStreamingContext`](api/streaming/index.html#org.apache.spark.streaming.api.java.JavaStreamingContext)
+object has to be created, which is the main entry point of all Spark Streaming functionality.
+A `JavaStreamingContext` object can be created by using
-# DStream Operations
-Data received from the input streams can be processed using _DStream operations_. There are two kinds of operations - _transformations_ and _output operations_. Similar to RDD transformations, DStream transformations operate on one or more DStreams to create new DStreams with transformed data. After applying a sequence of transformations to the input streams, output operations need to called, which write data out to an external data sink like a file system or a database.
+{% highlight scala %}
+new JavaStreamingContext(master, appName, batchInterval, [sparkHome], [jars])
+{% endhighlight %}
+
+
+ 
+
+ 
+
+
+
+Spark Streaming will monitor the directory `dataDirectory` for any Hadoop-compatible filesystem
+and process any files created in that directory. Note that
+
+ * The files must have the same data format.
+ * The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
+ the data directory.
+ * Once moved the files must not be changed.
+
+For more details on streams from files, Akka actors and sockets,
+see the API documentations of the relevant functions in
+[StreamingContext](api/streaming/index.html#org.apache.spark.streaming.StreamingContext) for
+Scala and [JavaStreamingContext](api/streaming/index.html#org.apache.spark.streaming.api.java.JavaStreamingContext)
+ for Java.
+
+Additional functionality for creating DStreams from sources such as Kafka, Flume, and Twitter
+can be imported by adding the right dependencies as explained in an
+[earlier](#linking) section. To take the
+case of Kafka, after adding the artifact `spark-streaming-kafka_{{site.SCALA_BINARY_VERSION}}` to the
+project dependencies, you can create a DStream from Kafka as
+
+
+{% highlight scala %}
+streamingContext.fileStream(dataDirectory)
+{% endhighlight %}
+
+
+{% highlight java %}
+javaStreamingContext.fileStream(dataDirectory);
+{% endhighlight %}
+
+
+
-## Transformations
+For more details on these additional sources, see the corresponding [API documentation]
+(#where-to-go-from-here). Furthermore, you can also implement your own custom receiver
+for your sources. See the [Custom Receiver Guide](streaming-custom-receivers.html).
-DStreams support many of the transformations available on normal Spark RDD's:
+## Operations
+There are two kinds of DStream operations - _transformations_ and _output operations_. Similar to
+RDD transformations, DStream transformations operate on one or more DStreams to create new DStreams
+with transformed data. After applying a sequence of transformations to the input streams, output
+operations need to called, which write data out to an external data sink, such as a filesystem or a
+database.
+
+### Transformations
+DStreams support many of the transformations available on normal Spark RDD's. Some of the
+common ones are as follows.
+{% highlight scala %}
+import org.apache.spark.streaming.kafka._
+KafkaUtils.createStream(streamingContext, kafkaParams, ...)
+{% endhighlight %}
+
+
+{% highlight java %}
+import org.apache.spark.streaming.kafka.*
+KafkaUtils.createStream(javaStreamingContext, kafkaParams, ...);
+{% endhighlight %}
+
+| Transformation | Meaning | ||
|---|---|---|---|
| Transformation | Meaning | ||
| map(func) | -Returns a new DStream formed by passing each element of the source DStream through a function func. | -||
| filter(func) | -Returns a new DStream formed by selecting those elements of the source DStream on which func returns true. | +Return a new DStream by passing each element of the source DStream through a + function func. | |
| flatMap(func) | - Similar to map, but each input item can be mapped to 0 or more output items (so func should return a Seq rather than a single item). |
+ Similar to map, but each input item can be mapped to 0 or more output items. | |
| mapPartitions(func) | -Similar to map, but runs separately on each partition (block) of the DStream, so func must be of type - Iterator[T] => Iterator[U] when running on an DStream of type T. | +filter(func) | +Return a new DStream by selecting only the records of the source DStream on which + func returns true. |
| repartition(numPartitions) | @@ -82,329 +465,681 @@ DStreams support many of the transformations available on normal Spark RDD's:|||
| union(otherStream) | -Return a new DStream that contains the union of the elements in the source DStream and the argument DStream. | +Return a new DStream that contains the union of the elements in the source DStream and + otherDStream. | |
| count() | -Returns a new DStream of single-element RDDs by counting the number of elements in each RDD of the source DStream. | +Return a new DStream of single-element RDDs by counting the number of elements in each RDD + of the source DStream. | |
| reduce(func) | -Returns a new DStream of single-element RDDs by aggregating the elements in each RDD of the source DStream using a function func (which takes two arguments and returns one). The function should be associative so that it can be computed in parallel. | +Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the + source DStream using a function func (which takes two arguments and returns one). + The function should be associative so that it can be computed in parallel. | |
| countByValue() | -When called on a DStream of elements of type K, returns a new DStream of (K, Long) pairs where the value of each key is its frequency in each RDD of the source DStream. | -||
| groupByKey([numTasks]) | - When called on a DStream of (K, V) pairs, returns a new DStream of (K, Seq[V]) pairs by grouping together all the values of each key in the RDDs of the source DStream. - Note: By default, this uses Spark's default number of parallel tasks (2 for local machine, 8 for a cluster) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks.
- |
+ When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs + where the value of each key is its frequency in each RDD of the source DStream. | |
| reduceByKey(func, [numTasks]) | - When called on a DStream of (K, V) pairs, returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument. |
+ When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the
+ values for each key are aggregated using the given reduce function. Note: By default,
+ this uses Spark's default number of parallel tasks (2 for local machine, 8 for a cluster) to
+ do the grouping. You can pass an optional numTasks argument to set a different
+ number of tasks. |
|
| join(otherStream, [numTasks]) | -When called on two DStreams of (K, V) and (K, W) pairs, returns a new DStream of (K, (V, W)) pairs with all pairs of elements for each key. | +When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) + pairs with all pairs of elements for each key. | |
| cogroup(otherStream, [numTasks]) | -When called on DStream of (K, V) and (K, W) pairs, returns a new DStream of (K, Seq[V], Seq[W]) tuples. | +When called on DStream of (K, V) and (K, W) pairs, return a new DStream of + (K, Seq[V], Seq[W]) tuples. | |
| transform(func) | -Returns a new DStream by applying func (a RDD-to-RDD function) to every RDD of the stream. This can be used to do arbitrary RDD operations on the DStream. | +Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. + This can be used to do arbitrary RDD operations on the DStream. | |
| updateStateByKey(func) | -Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values of each key. This can be used to track session state by using the session-id as the key and updating the session state as new data is received. | -
| Transformation | Meaning | |
|---|---|---|
| window(windowDuration, slideDuration) | -Return a new DStream which is computed based on windowed batches of the source DStream. windowDuration is the width of the window and slideTime is the frequency during which the window is calculated. Both times must be multiples of the batch interval. - | -|
| countByWindow(windowDuration, slideDuration) | - Return a sliding count of elements in the stream. windowDuration and slideDuration are exactly as defined in window().
- |
-|
| reduceByWindow(func, windowDuration, slideDuration) | - Return a new single-element stream, created by aggregating elements in the stream over a sliding interval using func. The function should be associative so that it can be computed correctly in parallel. windowDuration and slideDuration are exactly as defined in window().
- |
-|
| groupByKeyAndWindow(windowDuration, slideDuration, [numTasks]) - | - When called on a DStream of (K, V) pairs, returns a new DStream of (K, Seq[V]) pairs by grouping together values of each key over batches in a sliding window. -Note: By default, this uses Spark's default number of parallel tasks (2 for local machine, 8 for a cluster) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks. |
-|
| reduceByKeyAndWindow(func, windowDuration, slideDuration, [numTasks]) | - When called on a DStream of (K, V) pairs, returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function func over batches in a sliding window. Like in groupByKeyAndWindow, the number of reduce tasks is configurable through an optional second argument.
- windowDuration and slideDuration are exactly as defined in window().
- |
-|
| reduceByKeyAndWindow(func, invFunc, windowDuration, slideDuration, [numTasks]) | - A more efficient version of the above reduceByKeyAndWindow() where the reduce value of each window is calculated
- incrementally using the reduce values of the previous window. This is done by reducing the new data that enter the sliding window, and "inverse reducing" the old data that leave the window. An example would be that of "adding" and "subtracting" counts of keys as the window slides. However, it is applicable to only "invertible reduce functions", that is, those reduce functions which have a corresponding "inverse reduce" function (taken as parameter invFunc. Like in groupByKeyAndWindow, the number of reduce tasks is configurable through an optional second argument.
- windowDuration and slideDuration are exactly as defined in window().
- |
-|
| countByValueAndWindow(windowDuration, slideDuration, [numTasks]) | - When called on a DStream of (K, V) pairs, returns a new DStream of (K, Long) pairs where the value of each key is its frequency within a sliding window. Like in groupByKeyAndWindow, the number of reduce tasks is configurable through an optional second argument.
- windowDuration and slideDuration are exactly as defined in window().
- |
+ Return a new "state" DStream where the state for each key is updated by applying the + given function on the previous state of the key and the new values for the key. This can be + used to maintain arbitrary state data for each ket. |
UpdateStateByKey Operation
-| Operator | Meaning |
|---|---|
| foreachRDD(func) | -The fundamental output operator. Applies a function, func, to each RDD generated from the stream. This function should have side effects, such as printing output, saving the RDD to external files, or writing it over the network to an external system. | -
| print() | -Prints first ten elements of every batch of data in a DStream on the driver. | -
| saveAsObjectFiles(prefix, [suffix]) | - Save this DStream's contents as a SequenceFile of serialized objects. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]".
- |
-
| saveAsTextFiles(prefix, [suffix]) | -Save this DStream's contents as a text files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". | -
| saveAsHadoopFiles(prefix, [suffix]) | -Save this DStream's contents as a Hadoop file. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". | -
+
+{% highlight java %}
+Function2, Optional, Optional> updateFunction =
+ new Function2, Optional, Optional>() {
+ @Override public Optional call(List values, Optional state) {
+ Integer newSum = ... // add the new values with the previous running count to get the new count
+ return Optional.of(newSum)
+ }
+ }
{% endhighlight %}
-# Example
-A simple example to start off is the [NetworkWordCount](https://github.com/apache/incubator-spark/tree/master/examples/src/main/scala/org/apache/spark/streaming/examples/NetworkWordCount.scala). This example counts the words received from a network server every second. Given below is the relevant sections of the source code. You can find the full source code in `/streaming/src/main/scala/org/apache/spark/streaming/examples/NetworkWordCount.scala` .
+This is applied on a DStream containing words (say, the `pairs` DStream containing `(word,
+1)` pairs in the [quick example](#a-quick-example)).
-{% highlight scala %}
-import org.apache.spark.streaming.{Seconds, StreamingContext}
-import StreamingContext._
-...
+{% highlight java %}
+JavaPairDStream runningCounts = pairs.updateStateByKey(updateFunction);
+{% endhighlight %}
-// Create the context and set up a network input stream to receive from a host:port
-val ssc = new StreamingContext(args(0), "NetworkWordCount", Seconds(1))
-val lines = ssc.socketTextStream(args(1), args(2).toInt)
+
+
-// Split the lines into words, count them, and print some of the counts on the master
-val words = lines.flatMap(_.split(" "))
-val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
-wordCounts.print()
+The update function will be called for each word, with `newValues` having a sequence of 1's (from
+the `(word, 1)` pairs) and the `runningCount` having the previous count. For the complete
+Scala code, take a look at the example
+[StatefulNetworkWordCount]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/scala/org/apache/spark/streaming/examples/StatefulNetworkWordCount.scala).
-// Start the computation
-ssc.start()
-{% endhighlight %}
+Transform Operation
-The `socketTextStream` returns a DStream of text data received from a TCP server socket. The `lines` DStream is _transformed_ into a DStream using the `flatMap` operation, where each line is split into words. This `words` DStream is then mapped to a DStream of `(word, 1)` pairs, which is finally reduced to get the word counts. `wordCounts.print()` will print 10 of the counts generated every second. +The `transform` operation (along with its variations like `transformWith`) allows +arbitrary RDD-to-RDD functions to be applied on a DStream. It can be used to apply any RDD +operation that is not exposed in the DStream API. +For example, the functionality of joining every batch in a data stream +with another dataset is not directly exposed in the DStream API. However, +you can easily use `transform` to do this. This enables very powerful possibilities. For example, +if you want to do real-time data cleaning by joining the input data stream with precomputed +spam information (maybe generated with Spark as well) and then filtering based on it. -To run this example on your local machine, you need to first run a Netcat server by using +
+
-
-{% highlight bash %}
-$ nc -lk 9999
+{% highlight scala %}
+val spamInfoRDD = sparkContext.hadoopFile(...) // RDD containing spam information
+
+val cleanedDStream = inputDStream.transform(rdd => {
+ rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
+ ...
+})
{% endhighlight %}
-Then, in a different terminal, you can start NetworkWordCount by using
+
+
-{% highlight bash %}
-$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999
+{% highlight java %}
+// RDD containing spam information
+JavaPairRDD spamInfoRDD = javaSparkContext.hadoopFile(...);
+
+JavaPairDStream cleanedDStream = inputDStream.transform(
+ new Function, JavaPairRDD>() {
+ @Override public JavaPairRDD call(JavaPairRDD rdd) throws Exception {
+ rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
+ ...
+ }
+ });
{% endhighlight %}
-This will make NetworkWordCount connect to the netcat server. Any lines typed in the terminal running the netcat server will be counted and printed on screen.
+
+
-{% highlight bash %}
-# TERMINAL 1
-# RUNNING NETCAT
+In fact, you can also use [machine learning](mllib-guide.html) and
+[graph computation](graphx-programming-guide.html) algorithms in the `transform` method.
-$ nc -lk 9999
-hello world
+Window Operations+Finally, Spark Streaming also provides *windowed computations*, which allow you to apply +transformations over a sliding window of data. This following figure illustrates this sliding +window. +
+
+
+
+{% highlight scala %}
+// Reduce last 30 seconds of data, every 10 seconds
+val windowedWordCounts = pairs.reduceByKeyAndWindow(_ + _, Seconds(30), Seconds(10))
{% endhighlight %}
- |
-
-{% highlight bash %}
-# TERMINAL 2: RUNNING NetworkWordCount
-...
--------------------------------------------
-Time: 1357008430000 ms
--------------------------------------------
-(hello,1)
-(world,1)
-...
+
+
+
+{% highlight java %}
+// Reduce function adding two integers, defined separately for clarity
+Function2 |
-
+| Transformation | Meaning |
|---|---|
| window(windowLength, slideInterval) | +Return a new DStream which is computed based on windowed batches of the source DStream. + | +
| countByWindow(windowLength, slideInterval) | +Return a sliding window count of elements in the stream. + | +
| reduceByWindow(func, windowLength, slideInterval) | +Return a new single-element stream, created by aggregating elements in the stream over a + sliding interval using func. The function should be associative so that it can be computed + correctly in parallel. + | +
| reduceByKeyAndWindow(func, windowLength, slideInterval, + [numTasks]) | + When called on a DStream of (K, V) pairs, returns a new DStream of (K, V)
+ pairs where the values for each key are aggregated using the given reduce function func
+ over batches in a sliding window. Note: By default, this uses Spark's default number of
+ parallel tasks (2 for local machine, 8 for a cluster) to do the grouping. You can pass an optional
+ numTasks argument to set a different number of tasks.
+ |
+
| reduceByKeyAndWindow(func, invFunc, windowLength, + slideInterval, [numTasks]) | + A more efficient version of the above reduceByKeyAndWindow() where the reduce
+ value of each window is calculated incrementally using the reduce values of the previous window.
+ This is done by reducing the new data that enter the sliding window, and "inverse reducing" the
+ old data that leave the window. An example would be that of "adding" and "subtracting" counts
+ of keys as the window slides. However, it is applicable to only "invertible reduce functions",
+ that is, those reduce functions which have a corresponding "inverse reduce" function (taken as
+ parameter invFunc. Like in reduceByKeyAndWindow, the number of reduce tasks
+ is configurable through an optional argument.
+ |
+
| countByValueAndWindow(windowLength, + slideInterval, [numTasks]) | + When called on a DStream of (K, V) pairs, returns a new DStream of (K, Long) pairs where the
+ value of each key is its frequency within a sliding window. Like in
+ reduceByKeyAndWindow, the number of reduce tasks is configurable through an
+ optional argument.
+ |
+
| Output Operation | Meaning |
|---|---|
| print() | +Prints first ten elements of every batch of data in a DStream on the driver. | +
| foreachRDD(func) | +The fundamental output operator. Applies a function, func, to each RDD generated from + the stream. This function should have side effects, such as printing output, saving the RDD to + external files, or writing it over the network to an external system. | +
| saveAsObjectFiles(prefix, [suffix]) | + Save this DStream's contents as a SequenceFile of serialized objects. The file
+ name at each batch interval is generated based on prefix and
+ suffix: "prefix-TIME_IN_MS[.suffix]".
+ |
+
| saveAsTextFiles(prefix, [suffix]) | +Save this DStream's contents as a text files. The file name at each batch interval is + generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". | +
| saveAsHadoopFiles(prefix, [suffix]) | +Save this DStream's contents as a Hadoop file. The file name at each batch interval is + generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". | +
-
-
- Reducing the processing time of each batch of data by efficiently using cluster resources. -
- Setting the right batch size such that the data processing can keep up with the data ingestion. +
- + Reducing the processing time of each batch of data by efficiently using cluster resources. + +
- + Setting the right batch size such that the data processing can keep up with the data ingestion. +
+
+
+**Note**: If Spark Streaming and/or the Spark Streaming program is recompiled,
+you *must* create a new `StreamingContext` or `JavaStreamingContext`,
+not recreate from checkpoint data. This is because trying to load a
+context from checkpoint data may fail if the data was generated before recompilation of the
+classes. So, if you are using `getOrCreate`, then make sure that the checkpoint directory is
+explicitly deleted every time recompiled code needs to be launched.
+
+This failure recovery can be done automatically using Spark's
+[standalone cluster mode](spark-standalone.html), which allows any Spark
+application's driver to be as well as, ensures automatic restart of the driver on failure (see
+[supervise mode](spark-standalone.html#launching-applications-inside-the-cluster)). This can be
+tested locally by launching the above example using the supervise mode in a
+local standalone cluster and killing the java process running the driver (will be shown as
+*DriverWrapper* when `jps` is run to show all active Java processes). The driver should be
+automatically restarted, and the word counts will cont
+
+For other deployment environments like Mesos and Yarn, you have to restart the driver through other
+mechanisms.
+
+
-All this is periodically saved in the checkpoint directory. To recover, a new `StreamingContext` can be created with this directory by using
+This behavior is made simple by using `StreamingContext.getOrCreate`. This is used as follows.
{% highlight scala %}
-val ssc = new StreamingContext(checkpointDirectory)
+// Function to create and setup a new StreamingContext
+def functionToCreateContext(): StreamingContext = {
+ val ssc = new StreamingContext(...) // new context
+ val lines = ssc.socketTextStream(...) // create DStreams
+ ...
+ ssc.checkpoint(checkpointDirectory) // set checkpoint directory
+ ssc
+}
+
+// Get StreaminContext from checkpoint data or create a new one
+val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)
+
+// Do additional setup on context that needs to be done,
+// irrespective of whether it is being started or restarted
+context. ...
+
+// Start the context
+context.start()
+context.awaitTermination()
{% endhighlight %}
-On calling `ssc.start()` on this new context, the following steps are taken by the system
+If the `checkpointDirectory` exists, then the context will be recreated from the checkpoint data.
+If the directory does not exist (i.e., running for the first time),
+then the function `functionToCreateContext` will be called to create a new
+context and set up the DStreams. See the Scala example
+[RecoverableNetworkWordCount]({{site.SPARK_GITHUB_URL}}/tree/master/examples/src/main/scala/org/apache/spark/streaming/examples/RecoverableNetworkWordCount.scala).
+This example appends the word counts of network data into a file.
-1. Schedule the transformations and output operations for all the time steps between the time when the driver failed and when it last checkpointed. This is also done for those time steps that were previously scheduled but not processed due to the failure. This will make the system recompute all the intermediate data from the checkpointed RDD files, etc.
-1. Restart the network receivers, if any, and continue receiving new data.
+You can also explicitly create a `StreamingContext` from the checkpoint data and start the
+ computation by using `new StreamingContext(checkpointDirectory)`.
-There are two different failure behaviors based on which input sources are used.
+
+
-1. _Using HDFS files as input source_ - Since the data is reliably stored on HDFS, all data can re-computed and therefore no data will be lost due to any failure.
-1. _Using any input source that receives data through a network_ - The received input data is replicated in memory to multiple nodes. Since, all the data in the Spark worker's memory is lost when the Spark driver fails, the past input data will not be accessible and driver recovers. Hence, if stateful and window-based operations are used (like `updateStateByKey`, `window`, `countByValueAndWindow`, etc.), then the intermediate state will not be recovered completely.
+This behavior is made simple by using `JavaStreamingContext.getOrCreate`. This is used as follows.
-In future releases, we will support full recoverability for all input sources. Note that for non-stateful transformations like `map`, `count`, and `reduceByKey`, with _all_ input streams, the system, upon restarting, will continue to receive and process new data.
+{% highlight java %}
+// Create a factory object that can create a and setup a new JavaStreamingContext
+JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {
+ JavaStreamingContextFactory create() {
+ JavaStreamingContext jssc = new JavaStreamingContext(...); // new context
+ JavaDStream lines = jssc.socketTextStream(...); // create DStreams
+ ...
+ jssc.checkpoint(checkpointDirectory); // set checkpoint directory
+ return jssc;
+ }
+};
+
+// Get JavaStreamingContext from checkpoint data or create a new one
+JavaStreamingContext context = JavaStreamingContext.getOrCreate(checkpointDirectory, contextFactory);
+
+// Do additional setup on context that needs to be done,
+// irrespective of whether it is being started or restarted
+context. ...
+
+// Start the context
+context.start();
+context.awaitTermination();
+{% endhighlight %}
-To better understand the behavior of the system under driver failure with a HDFS source, lets consider what will happen with a file input stream Specifically, in the case of the file input stream, it will correctly identify new files that were created while the driver was down and process them in the same way as it would have if the driver had not failed. To explain further in the case of file input stream, we shall use an example. Lets say, files are being generated every second, and a Spark Streaming program reads every new file and output the number of lines in the file. This is what the sequence of outputs would be with and without a driver failure.
+If the `checkpointDirectory` exists, then the context will be recreated from the checkpoint data.
+If the directory does not exist (i.e., running for the first time),
+then the function `contextFactory` will be called to create a new
+context and set up the DStreams. See the Scala example
+[JavaRecoverableWordCount]({{site.SPARK_GITHUB_URL}}/tree/master/examples/src/main/scala/org/apache/spark/streaming/examples/JavaRecoverableWordCount.scala)
+(note that this example is missing in the 0.9 release, so you can test it using the master branch).
+This example appends the word counts of network data into a file.
+
+You can also explicitly create a `JavaStreamingContext` from the checkpoint data and start
+the computation by using `new JavaStreamingContext(checkpointDirectory)`.
+
+
+