Unit 3: Using RDD for Creating Applications in Spark and Graph Analytics

I. Foundations of Distributed Data Processing

Apache Spark is an open-source distributed computing engine developed at UC Berkeley’s AMPLab (2009). Its core abstraction, the Resilient Distributed Dataset (RDD), represents data partitioned across cluster nodes and processed through parallel, fault-tolerant operations.

• Governing principle: Spark sends computation to distributed data partitions, reducing unnecessary data movement.
• Cluster roles:
  • Driver: Runs the application, creates the SparkContext, builds execution plans, and schedules work.
  • Executor: Runs tasks and stores cached partitions on worker nodes.
  • Cluster manager: Allocates resources through Standalone, YARN, Kubernetes, or another supported manager.
• Lazy evaluation: Transformations construct a directed acyclic graph (DAG); an action triggers its execution.
• Partitioned processing: Each RDD contains logical partitions that Spark can process concurrently.
• Fault tolerance: Lost partitions are reconstructed from the RDD’s lineage rather than restored only from replicas.
• Functional convention: Operations generally create new datasets instead of modifying existing RDDs.

II. Resilient Distributed Datasets — Spark’s Core Data Abstraction

An RDD is an immutable, distributed collection of records that can be processed in parallel. “Resilient” refers to lineage-based recovery, while “distributed dataset” indicates that records are divided into partitions across a cluster.

A. Features of RDD

RDDs provide explicit control over distributed data, partitioning, persistence, and low-level functional transformations.

• Immutability: Applying map to RDD[Int] produces another RDD; the source RDD remains unchanged.
• In-memory computation: Reused data can be cached with cache() or persist(), accelerating iterative algorithms.
• Fault tolerance: If a partition of mapped = source.map(f) is lost, Spark reruns f on the corresponding source partition.
• Lineage: rdd.toDebugString exposes dependencies used to reconstruct partitions.
• Partitioning: rdd.getNumPartitions reports parallel divisions; pair RDDs may use HashPartitioner or RangePartitioner.
• Type support: RDDs can hold Scala, Java, Python, or user-defined objects, subject to serialization requirements.
• Location awareness: Spark schedules tasks near their input blocks when possible.
• Coarse-grained operations: RDD APIs transform collections or partitions, rather than performing arbitrary updates to individual records.

B. Creating RDDs

RDDs are created by parallelizing local collections, reading external data, or transforming existing RDDs.

• Local collection: parallelize divides driver-side data into partitions.

val numbers = sc.parallelize(Seq(1, 2, 3, 4), 2)

• sc is the SparkContext.
• 2 is the requested number of partitions.
  • External storage: textFile creates an RDD[String], usually with one record per line.

val lines = sc.textFile("hdfs:///data/logs.txt")

• Existing RDD: Transformations derive new RDDs while retaining lineage.

val errors = lines.filter(_.contains("ERROR"))

• Sequence files and Hadoop inputs: APIs such as sequenceFile and newAPIHadoopRDD connect RDDs to Hadoop-compatible formats.
• DataFrame conversion: dataFrame.rdd exposes rows as RDD[Row], although structured APIs are normally preferable for SQL-style processing.

C. RDD functions

Functions supplied to RDD transformations define record-level or partition-level computation and should avoid unsafe external side effects.

• Element function: map(x => x * 2) applies one function to every element.
• Predicate function: filter(x => x > 0) retains records for which the Boolean result is true.
• One-to-many function: flatMap(_.split(" ")) emits zero or more output elements for each input.
• Binary reduction function: reduce((a, b) => a + b) combines records associatively.
• Pair function: map(word => (word, 1)) forms key-value records for aggregation.
• Partition function: mapPartitions(iterator => ...) initializes expensive resources once per partition.
• Closure rule: Spark serializes referenced variables and sends their captured values to executors; changing a driver variable inside a task does not reliably update the driver’s copy.
• Design requirement: Reduction functions should be associative and commutative because partition order is not guaranteed.

D. RDD operations and methods

RDD operations are divided into lazy transformations and execution-triggering actions.

1. Transformations create RDDs and record dependencies.
   • Narrow transformations: map, filter, and flatMap let each output partition depend on a small number of input partitions.
   • Wide transformations: groupByKey, reduceByKey, join, and distinct may require a shuffle across executors.
   • Set-like methods: union, intersection, subtract, and cartesian combine RDDs with different costs.
2. Actions compute results or write output.
   • Driver results: count, first, take(n), reduce, and collect.
   • Storage results: saveAsTextFile and saveAsSequenceFile.
   • Side-effect traversal: foreach executes a function on executor-side records.

• Aggregation choice: reduceByKey(_ + _) performs local combining before shuffle and is generally more efficient than groupByKey.
• Persistence methods: cache() uses the default storage level; persist(level) selects memory, disk, or serialized storage.
• Lifecycle method: unpersist() releases cached blocks when reuse ends.
• Safety constraint: collect() places every result on the driver and can exhaust driver memory; take(n) is safer for inspection.

III. Spark Application Environment — Interactive Execution and Shared State

Spark applications require an initialized driver environment and controlled mechanisms for distributing read-only data or collecting aggregated task information.

A. Invoking the Spark shell

The Spark shell provides an interactive environment in which Spark initializes its core session objects automatically.

• Scala shell: Run spark-shell; it normally creates sc: SparkContext and spark: SparkSession.
• Python shell: Run pyspark to obtain corresponding Python objects.
• Local execution:

spark-shell --master local[4]

• local[4] uses four local worker threads.
  • Cluster execution:

spark-shell --master yarn --executor-memory 4G

• --master yarn requests YARN resources.
• --executor-memory 4G assigns four gibibytes to each executor process.
  • Package loading: --packages group:artifact:version resolves additional libraries before startup.
  • Basic inspection:

val data = sc.parallelize(1 to 10)
data.filter(_ % 2 == 0).collect()

• Operational role: The shell suits exploration and debugging; production applications are normally packaged and submitted with spark-submit.

B. Shared variables

Spark provides broadcast variables and accumulators because ordinary closure variables are copied to executors rather than consistently shared.

1. Broadcast variables distribute a large read-only value once per executor for efficient reuse.

val rates = sc.broadcast(Map("USD" -> 1.0, "EUR" -> 1.08))
val usd = prices.map { case (currency, value) =>
  value * rates.value(currency)
}

• Access: Executors read the value through rates.value.
• Use case: Lookup tables, model parameters, and configuration maps.
• Cleanup: unpersist() removes executor copies; destroy() invalidates the broadcast permanently.
  2. Accumulators support associative additions from tasks while the driver reads the result.

val invalid = sc.longAccumulator("invalid-records")
records.foreach(r => if (!valid(r)) invalid.add(1))
println(invalid.value)

• Use case: Counters and diagnostic sums.
• Reliability constraint: Task retries can complicate updates inside transformations; accumulator values should not control application correctness.

IV. Spark GraphX — Distributed Graph Processing

GraphX is Spark’s graph-processing API for Scala and Java. It represents a directed multigraph as vertices and edges while integrating graph computation with RDD-based processing.

A. Introduction to Spark GraphX

A GraphX graph combines a vertex RDD with an edge RDD and assigns attributes to both.

• Vertex identity: A vertex uses a 64-bit VertexId, which is an alias for Long.
• Vertex representation: VertexRDD[VD] stores pairs such as (1L, "Alice"), where VD is the vertex-attribute type.
• Edge representation: EdgeRDD[ED] stores Edge(srcId, dstId, attr), where ED is the edge-attribute type.
• Graph construction:

val vertices = sc.parallelize(Seq((1L, "A"), (2L, "B")))
val edges = sc.parallelize(Seq(Edge(1L, 2L, 5)))
val graph = Graph(vertices, edges)

• Triplet view: An EdgeTriplet exposes the source attribute, edge attribute, and destination attribute together.
• Directed semantics: Edge direction matters for operations such as in-degree, out-degree, and PageRank.

B. Spark GraphX features

GraphX supplies graph-parallel abstractions, structural operators, and standard algorithms on Spark’s fault-tolerant runtime.

• Unified processing: Graph data can be transformed using both graph operators and RDD methods.
• Property graphs: Vertices and edges carry arbitrary attributes, such as user names and relationship weights.
• Vertex-cut partitioning: GraphX can distribute edges while replicating frequently referenced vertices to reduce communication.
• Structural views: graph.vertices, graph.edges, and graph.triplets expose graph components.
• Built-in algorithms: The library includes PageRank, connected components, triangle counting, and label propagation.
• Pregel API: pregel expresses iterative, message-driven graph computation through vertex programs, message sending, and message merging.
• Fault recovery: Graph transformations inherit Spark lineage and persistence facilities.
• Language limitation: GraphX’s primary API is JVM-based and is not a native Python graph API.

C. Spark GraphX operations

GraphX operations modify properties, inspect structure, join external data, or aggregate information around vertices.

• Property transformation: mapVertices changes vertex attributes, while mapEdges changes edge attributes without altering topology.
• Filtering: subgraph selects vertices and edges using predicates; mask restricts one graph to another graph’s structure.
• Degree calculation: inDegrees, outDegrees, and degrees return vertex-degree values.
• Joining: joinVertices and outerJoinVertices merge external vertex data by VertexId.
• Neighborhood aggregation: aggregateMessages sends values along edges and combines messages arriving at each vertex.
• Reversal: reverse exchanges every edge’s source and destination while retaining attributes.
• Algorithm invocation:

val ranks = graph.pageRank(0.0001).vertices

• 0.0001 is the convergence tolerance.
• The result associates each VertexId with a PageRank score.
  • Performance control: Frequently reused graphs should be cached, and an appropriate partition strategy should be selected for the graph’s structure.

V. Feature Engineering — Preparing Data for Analytical Models

Feature engineering converts raw records into numerical representations that machine-learning algorithms can process consistently.

A. Feature extraction and transformation

Feature extraction derives measurable variables, while transformation changes their scale, dimensionality, or representation.

1. Extraction creates features from raw input.
   • Text frequency: HashingTF maps terms into a fixed-length vector using feature hashing.
   • TF-IDF: Term frequency measures within-document occurrence; inverse document frequency reduces the influence of common terms:

IDF(t) = log((N + 1) / (DF(t) + 1))
TFIDF(t,d) = TF(t,d) x IDF(t)

• N is the number of documents, DF(t) the documents containing term t, and TF(t,d) the frequency of t in document d.
• Categorical input: Indexing and one-hot encoding represent categories numerically without treating category labels as continuous magnitudes.
  2. Transformation produces a model-suitable feature space.
• Scaling: StandardScaler can center features and scale them using standard deviation.
• Normalization: Normalizer rescales each vector to a chosen norm, commonly L1 or L2.
• Dimensionality reduction: Principal component analysis projects vectors onto fewer orthogonal components.
• Vector assembly: VectorAssembler combines numeric columns into one feature vector in the DataFrame-based ML API.

• Pipeline discipline: Parameters such as IDF values, category indexes, or scaling statistics must be fitted on training data and then applied unchanged to validation and test data.
• Sparse representation: High-dimensional text and categorical features should use sparse vectors when most entries are zero.
• RDD context: Legacy spark.mllib supports RDD-based feature utilities, while spark.ml DataFrame pipelines are preferred for current application development.