import numpy as np

data = np.arange(1000).reshape(100,10)
print data.shape

"""
Explanation: Simple Aggregation
End of explanation
"""


import pandas as pd

pand_tmp = pd.DataFrame(data, 
             columns=['x{0}'.format(i) for i in range(data.shape[1])])
pand_tmp.head()

"""
Explanation: Pandas
End of explanation
"""


pand_tmp.sum(axis=1)

"""
Explanation: What is the row sum?
End of explanation
"""


pand_tmp.sum(axis=0)

pand_tmp.to_csv('numbers.csv', index=False)

"""
Explanation: Column sum?
End of explanation
"""


import findspark
import os
findspark.init() # you need that before import pyspark.

import pyspark
sc = pyspark.SparkContext('local[4]', 'pyspark')

lines = sc.textFile('numbers.csv', 18)
for l in lines.take(3):
    print l

lines.take(3)

type(lines.take(1))

"""
Explanation: Spark
End of explanation
"""


lines = lines.filter(lambda x: x.find('x') != 0)
for l in lines.take(2):
    print l

data = lines.map(lambda x: x.split(','))
data.take(3)

"""
Explanation: How do we skip the header?  How about using find()?  What is Boolean value for true with find()?
End of explanation
"""


def row_sum(x):
    int_x = map(lambda x: int(x), x)
    return sum(int_x)

data_row_sum = data.map(row_sum)

print data_row_sum.collect()

print data_row_sum.count()

"""
Explanation: Row Sum
Cast to integer and sum!
End of explanation
"""


def col_key(x):
    for i, value in enumerate(x):
        yield (i, int(value))

tmp = data.flatMap(col_key)
tmp.take(15)

"""
Explanation: Column Sum
This one's a bit trickier, and portends ill for large, complex data sets (like example 5)...
Let's enumerate the list comprising each RDD "line" such that each value is indexed by the corresponding column number.
End of explanation
"""


tmp.take(3)

tmp = tmp.groupByKey()
for i in tmp.take(2):
    print i, type(i)

data_col_sum = tmp.map(lambda x: sum(x[1]))
for i in data_col_sum.take(2):
    print i

print data_col_sum.collect()
print data_col_sum.count()

"""
Explanation: Notice how flatMap works here: the generator is returned per partition, meaning that the first element value of each tuple cycles.
End of explanation
"""


from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

sc

pyspark_df = sqlContext.createDataFrame(pand_tmp)

pyspark_df.take(2)

"""
Explanation: Column sum with Spark.sql.dataframe
End of explanation
"""


for i in pyspark_df.columns:
    print pyspark_df.groupBy().sum(i).collect()

"""
Explanation: groupBy() without arguments groups by all columns
End of explanation
"""


