def svd(blocked_rdd, k):
<EOL>c = blocked_rdd.map(lambda x: (x.T.dot(x), x.shape[<NUM_LIT:0>]))<EOL>prod, n = c.reduce(lambda x, y: (x[<NUM_LIT:0>] + y[<NUM_LIT:0>], x[<NUM_LIT:1>] + y[<NUM_LIT:1>]))<EOL>w, v = ln.eig(prod / n)<EOL>w = np.real(w)<EOL>v = np.real(v)<EOL>inds = np.argsort(w)[::-<NUM_LIT:1>]<EOL>s = np.sqrt(w[inds[<NUM_LIT:0>:k]]) * np.sqrt(n)<EOL>v = v[:, inds[<NUM_LIT:0>:k]].T<EOL>u = blocked_rdd.map(lambda x: np.inner(x, v) / s)<EOL>return u, s, v<EOL>
Calculate the SVD of a blocked RDD directly, returning only the leading k
singular vectors. Assumes n rows and d columns, efficient when n >> d
Must be able to fit d^2 within the memory of a single machine.
Parameters
----------
blocked_rdd : RDD
    RDD with data points in numpy array blocks
k : Int
    Number of singular vectors to return
Returns
----------
u : RDD of blocks
    Left eigenvectors
s : numpy array
    Singular values
v : numpy array
    Right eigenvectors