how to implement eigenvalue calculation with MapReduce/Hadoop?

MapReduce

Hadoop

Eigenvalue Calculation

Big Data

Distributed Computing

how to implement eigenvalue calculation with MapReduce/Hadoop?

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

Distributed Systems Big Data

Introduction

Computing eigenvalues of large matrices using MapReduce involves implementing iterative algorithms (Power Iteration, Lanczos, or distributed SVD) where each iteration's matrix-vector multiplication is parallelized across the cluster. The Map phase computes partial products of matrix rows with the current vector, and the Reduce phase aggregates them into the updated vector. Practically, libraries like Apache Mahout, Apache Spark MLlib, or custom Hadoop jobs handle this. Power Iteration is the simplest algorithm to distribute.

Power Iteration Algorithm

The power iteration method finds the dominant eigenvalue (largest in absolute value) and its eigenvector by repeatedly multiplying a matrix by a vector:

v_{k+1} = A * v_k / ||A * v_k||
eigenvalue ≈ v_k^T * A * v_k

Each iteration requires one matrix-vector multiplication, which is the operation we parallelize with MapReduce.

MapReduce Implementation

Input Format

The matrix A is stored as rows in HDFS, one row per line:

1# matrix.txt (row_index, col_index, value)
20,0,2.0
30,1,1.0
41,0,1.0
51,1,3.0

The current vector v is stored in a separate file or distributed via Hadoop's DistributedCache.

Mapper: Partial Matrix-Vector Products

python

1#!/usr/bin/env python3
2# mapper.py — computes partial products for each row
3
4import sys
5import json
6
7# Load current vector from distributed cache
8with open("current_vector.json", "r") as f:
9    vector = json.load(f)  # {index: value}
10
11for line in sys.stdin:
12    line = line.strip()
13    if not line:
14        continue
15
16    row_idx, col_idx, value = line.split(",")
17    row_idx = int(row_idx)
18    col_idx = int(col_idx)
19    value = float(value)
20
21    # Partial product: A[row][col] * v[col]
22    if str(col_idx) in vector:
23        partial = value * vector[str(col_idx)]
24        # Emit (row_index, partial_product)
25        print(f"{row_idx}\t{partial}")

Reducer: Sum Partial Products

python

1#!/usr/bin/env python3
2# reducer.py — sums partial products per row to get (Av)[row]
3
4import sys
5
6current_row = None
7row_sum = 0.0
8
9for line in sys.stdin:
10    line = line.strip()
11    row_idx, value = line.split("\t")
12    row_idx = int(row_idx)
13    value = float(value)
14
15    if current_row == row_idx:
16        row_sum += value
17    else:
18        if current_row is not None:
19            print(f"{current_row}\t{row_sum}")
20        current_row = row_idx
21        row_sum = value
22
23# Emit the last row
24if current_row is not None:
25    print(f"{current_row}\t{row_sum}")

Driver Script: Iterative Execution

python

1#!/usr/bin/env python3
2# driver.py — runs power iteration across multiple MapReduce jobs
3
4import subprocess
5import json
6import math
7
8def run_mapreduce_iteration(iteration, matrix_path, vector_path, output_path):
9    """Run one MapReduce job for matrix-vector multiplication."""
10    cmd = [
11        "hadoop", "jar", "/path/to/hadoop-streaming.jar",
12        "-files", f"mapper.py,reducer.py,{vector_path}",
13        "-mapper", "python3 mapper.py",
14        "-reducer", "python3 reducer.py",
15        "-input", matrix_path,
16        "-output", f"{output_path}/iter_{iteration}",
17    ]
18    subprocess.run(cmd, check=True)
19
20def read_vector_from_hdfs(path):
21    """Read the output vector from HDFS."""
22    result = subprocess.run(
23        ["hdfs", "dfs", "-cat", f"{path}/part-*"],
24        capture_output=True, text=True
25    )
26    vector = {}
27    for line in result.stdout.strip().split("\n"):
28        idx, val = line.split("\t")
29        vector[idx] = float(val)
30    return vector
31
32def normalize(vector):
33    """Normalize vector to unit length."""
34    norm = math.sqrt(sum(v ** 2 for v in vector.values()))
35    return {k: v / norm for k, v in vector.items()}, norm
36
37# Initialize random vector
38n = 1000  # Matrix dimension
39import random
40vector = {str(i): random.gauss(0, 1) for i in range(n)}
41vector, _ = normalize(vector)
42
43# Power iteration
44for iteration in range(50):
45    # Write current vector to HDFS
46    vector_path = f"/tmp/vector_iter_{iteration}.json"
47    with open(f"current_vector.json", "w") as f:
48        json.dump(vector, f)
49
50    # Run MapReduce
51    run_mapreduce_iteration(iteration, "/data/matrix.txt", vector_path, "/output/eigen")
52
53    # Read result and normalize
54    new_vector = read_vector_from_hdfs(f"/output/eigen/iter_{iteration}")
55    new_vector, eigenvalue = normalize(new_vector)
56
57    # Check convergence
58    diff = sum((new_vector.get(k, 0) - vector.get(k, 0)) ** 2 for k in vector)
59    print(f"Iteration {iteration}: eigenvalue ≈ {eigenvalue:.6f}, diff = {diff:.2e}")
60
61    if diff < 1e-10:
62        print(f"Converged after {iteration + 1} iterations")
63        break
64
65    vector = new_vector

Using Apache Spark (Modern Alternative)

Spark's MLlib provides distributed SVD and PCA without writing raw MapReduce:

python

1from pyspark.sql import SparkSession
2from pyspark.mllib.linalg.distributed import RowMatrix
3from pyspark.mllib.linalg import Vectors
4
5spark = SparkSession.builder.appName("Eigenvalues").getOrCreate()
6sc = spark.sparkContext
7
8# Create distributed matrix from rows
9rows = sc.parallelize([
10    Vectors.dense([2.0, 1.0, 0.0]),
11    Vectors.dense([1.0, 3.0, 1.0]),
12    Vectors.dense([0.0, 1.0, 2.0]),
13])
14
15matrix = RowMatrix(rows)
16
17# Compute SVD — singular values are related to eigenvalues
18svd = matrix.computeSVD(3, computeU=True)
19print("Singular values:", svd.s)
20# For symmetric matrices: eigenvalues = singular_values^2 / n
21
22# Or use PCA
23from pyspark.ml.feature import PCA
24from pyspark.ml.linalg import Vectors as MLVectors
25
26# PCA gives top-k eigenvalues via explained variance
27pca = PCA(k=2, inputCol="features", outputCol="pca_features")

Apache Mahout

bash

1# Mahout provides distributed eigenvalue decomposition
2mahout ssvd \
3  --input /data/matrix \
4  --output /output/svd \
5  --rank 10 \
6  --oversampling 20 \
7  --blockHeight 10000 \
8  --tempDir /tmp/mahout

Lanczos Algorithm (Better for Sparse Matrices)

For sparse matrices, the Lanczos algorithm is more efficient than Power Iteration because it finds multiple eigenvalues simultaneously:

python

1# Conceptual Lanczos iteration (each step is a MapReduce job)
2# 1. Start with random vector q_1
3# 2. For k = 1, 2, ..., m:
4#    a. w = A * q_k              (MapReduce: matrix-vector multiply)
5#    b. alpha_k = q_k^T * w      (MapReduce: dot product)
6#    c. w = w - alpha_k * q_k - beta_{k-1} * q_{k-1}
7#    d. beta_k = ||w||
8#    e. q_{k+1} = w / beta_k
9# 3. Eigenvalues of tridiagonal matrix T (alpha, beta) ≈ eigenvalues of A

Common Pitfalls

Running too many MapReduce iterations: Each iteration launches a full Hadoop job with job scheduling overhead (30-60 seconds). Power Iteration can require 50-100 iterations. Use Spark (in-memory iterations) instead of Hadoop Streaming for iterative algorithms, reducing per-iteration overhead from seconds to milliseconds.
Floating-point precision across distributed nodes: Summing partial products in different orders across reducers introduces floating-point rounding differences. For high-precision eigenvalue computation, use Kahan summation or math.fsum() in the reducer to minimize accumulated error.
Not normalizing the vector between iterations: Without normalization, the vector values grow exponentially (if the eigenvalue > 1) or shrink to zero (if < 1), causing overflow or underflow. Always normalize to unit length after each matrix-vector multiplication.
Power iteration only finds one eigenvalue: Power Iteration converges to the dominant eigenvalue only. For multiple eigenvalues, use deflation (subtract the found eigenvalue's contribution) or the Lanczos/Arnoldi algorithm. Spark MLlib's SVD finds all top-k eigenvalues in one computation.
Storing the full matrix in memory: Large matrices (millions of rows/columns) do not fit in memory. Store the matrix in sparse format on HDFS and stream rows through the mapper. Each mapper only needs one row of the matrix and the full vector (which is much smaller).

Summary

Eigenvalue computation via MapReduce parallelizes the matrix-vector multiplication step across the cluster
Power Iteration is the simplest algorithm: Map computes partial row-vector products, Reduce sums them per row
Each iteration requires a full MapReduce job, making Hadoop Streaming slow for iterative algorithms
Use Apache Spark MLlib (computeSVD, PCA) for practical distributed eigenvalue computation
The Lanczos algorithm finds multiple eigenvalues efficiently for large sparse matrices
Always normalize the vector between iterations to prevent numerical overflow/underflow

System Design Fundamentals

Build a strong foundation in designing scalable, reliable distributed systems.

View the course

Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

how to implement eigenvalue calculation with MapReduce/Hadoop?

System Design practice on Codemia

Introduction

Power Iteration Algorithm

MapReduce Implementation

Input Format

Mapper: Partial Matrix-Vector Products

Reducer: Sum Partial Products

Driver Script: Iterative Execution

Using Apache Spark (Modern Alternative)

Apache Mahout

Lanczos Algorithm (Better for Sparse Matrices)

Common Pitfalls

Summary

Related reading

System Design Fundamentals

Track what you have practised

System Design practice on Codemia