sc = pyspark.SparkContext(appName="spark-notebook")
ss = SparkSession(sc)

myRDD = sc.textFile("file:///path/to/part3/numbers.txt", 10)

"""
Explanation: Intialize a spark instance-
End of explanation
"""


myRDD.getNumPartitions()

print myRDD.take(20) # get first 20 values

"""
Explanation: Get number of RDD partitions-
End of explanation
"""


def square(value):
    return int(value)**2

newRDD = myRDD.map(square)

print newRDD.collect() # get map results

subRDD = newRDD.map(lambda x: (x, 1) if x%2==0 else (x, 0)) # using lamda functions

"""
Explanation: We will define a square function for our map operation-
End of explanation
"""


print subRDD.collect()

"""
Explanation: The above map function maps generated two types of key value pairs- (even, 1) and (odd, 0)
End of explanation
"""


# your code here


print testRDD.take(10)

"""
Explanation: Now define a lambda function to generate (key, value) pairs where key = even or odd depending upon the input and 
value = input
End of explanation
"""


reduced = testRDD.reduceByKey(add)

print reduced.collect()

sc.stop()

sc = pyspark.SparkContext(appName="spark-notebook")

mat = np.array([])
with open("./numbers.txt", "r") as file:
    for line in file:
        mat = np.hstack((mat, np.array(int(line))))
        
mymat = mat[:6]

print mymat

"""
Explanation: There are two types of Reduce operations- reduceByKey() and reduce(). Check PySpark documentation for more details
End of explanation
"""


from pyspark.mllib.linalg import Vectors
sv = Vectors.sparse(6,[0, 1, 3, 4],[1, 2, 4, 5])

print type(sv)

"""
Explanation: In MLlib you can use Dense or Sparse matrices for computation.
Create a Sparse vector for MLlib using mat-
End of explanation
"""


from pyspark.mllib.regression import LabeledPoint
pos = LabeledPoint(1.0, mat) # label(Y) = 1 and data(X) = mymat
print pos

"""
Explanation: Labeled point
A labeled point is a local vector, either dense or sparse, associated with a label/response. In MLlib, labeled points are used in supervised learning algorithms. We use a double to store a label, so we can use labeled points in both regression and classification.
End of explanation
"""


from pyspark.mllib.linalg import Matrix, Matrices
dm = Matrices.dense(2,2,mat[7:11]) # 2x2 dense matrix
print dm 

"""
Explanation: Local matrix
A local matrix has integer-typed row and column indices and double-typed values, stored on a single machine.
End of explanation
"""


newmat = np.reshape(mat[:6], (2,3)) # 2x3 matrix
print newmat

"""
Explanation: Distributed matrix
A distributed matrix has long-typed row and column indices and double-typed values, stored distributively in one or more RDDs. It is very important to choose the right format to store large and distributed matrices. Converting a distributed matrix to a different format may require a global shuffle, which is quite expensive.
End of explanation
"""


from pyspark.mllib.linalg.distributed import RowMatrix

# Create an RDD of newmat
rows = sc.parallelize(newmat)
rowmat = RowMatrix(rows)
print rowmat.numRows(), rowmat.numCols()

"""
Explanation: Row matrix
End of explanation
"""


from pyspark.mllib.linalg.distributed import BlockMatrix


# Create an RDD of sub-matrix blocks.
blocks = sc.parallelize([((0, 0), Matrices.dense(2,2,mat[7:11])),
                         ((1, 0), Matrices.dense(2,2,mat[1:5]))])

# Create a BlockMatrix from an RDD of sub-matrix blocks.
mat = BlockMatrix(blocks, 2, 2) # [A | B]
print mat

sc.stop()

"""
Explanation: BlockMatrix
End of explanation
"""


