#Create a list of 10000 integers
data = range(1, 10001)

#It will just print out the function but not data itself
print(data)

#Let's use SparkContext and parallelize() function to create a RDD Dataset and spread it across the cluster partitions
ds_rdd = sc.parallelize(data, 5)
ds_rdd_1 = sc.parallelize([('a',1),('a',2),('b',4),('b',1),('c',3),('c',3)])
ds_rdd_2 = sc.parallelize([('a',[1,2,3,4]),('b',[5,6,7])])
ds_rdd_3 = sc.parallelize(range(1, 11))
ds_rdd_4 = sc.parallelize([('b',1),('c',3),('c',3)])

#Collect() action will trigger execution and then we can print 10 first element of our list
print(ds_rdd.collect()[:10])
print(ds_rdd_1.collect())
print(ds_rdd_2.collect())
print(ds_rdd_3.collect())

help(sc.parallelize)

#Returns number of partitions
print("Number of RDD partitions: ",ds_rdd.getNumPartitions())
#Returns number of instances
print("Number of instances: ",ds_rdd.count())
#Returns sum of instances
print("Sum of Rdd instances: ",ds_rdd.sum())
#Returns number of instances by key
print("Number of instances by key: ",ds_rdd_1.countByKey())
#Returns number of instances by value
print("Number of instances by value: ",ds_rdd_1.countByValue())
#Returns dictionary with key,value pairs. Some instances are removed if keys are duplicated
print("Dictionary with key,value pairs: ",ds_rdd_1.collectAsMap())

#Returns min. value of RDD elements
print("Min. value: ",ds_rdd.min())
#Returns max. value of RDD elements
print("Max. value: ",ds_rdd.max())
#Returns mean. value of RDD elements
print("Mean value: ",ds_rdd.mean())
#Returns standard deviation value of RDD elements
print("Stdev. value: ",ds_rdd.stdev())
#Returns varaiance value of RDD elements
print("Variance value: ",ds_rdd.variance())
#Returns histogram for specified number of bins of RDD elements
print("Histogram: ",ds_rdd.histogram(10))
#Returns overall stats of RDD elements
print("Summary stats.: ",ds_rdd.stats())

#Apply lambda function on each rdd element
print(ds_rdd.map(lambda x: x*2).collect()[:10])
#Apply lambda function on each rdd element and flatten result
print(ds_rdd_1.flatMap(lambda x:x).collect())
#Apply flatMap function on key,value pair without changing key value
print(ds_rdd_2.flatMapValues(lambda x:x).collect())
#create a function and apply it to all rdd elements
def print_el(x): print(x)
ds_rdd.foreach(print_el)

#return a list of all elements of rdd
print(ds_rdd.collect()[:10])
#return a given number of rdd elements
print(ds_rdd.take(10))
#return first element of rdd
print(ds_rdd.first())
#return a given number of top (last) elements
print(ds_rdd.top(5))
#return sampled subset of rdd
print(ds_rdd.sample(withReplacement=False,fraction=0.001, seed=1).collect())
#filter rdd basing on some condition
print(ds_rdd_1.filter(lambda x: 'b' in x).collect())
#return distinct rdd values
print(ds_rdd_1.distinct().collect())
#return list of rdd keys
print(ds_rdd_1.keys().collect())
#return list of rdd values
print(ds_rdd_1.values().collect())

#Merge/reduce rdd values by key
print(ds_rdd_1.reduceByKey(lambda x,y:x+y).collect())
#Merge/reduce rdd values
print(ds_rdd_1.reduce(lambda x,y:x+y))
#return rdd of grouped values
print(ds_rdd_3.groupBy(lambda x: x%2).mapValues(list).collect())
#group rdd by key
print(ds_rdd_1.groupByKey().mapValues(list).collect())

#returns all rdd values avaliable in one rdd and not available in the other one
print(ds_rdd_1.subtract(ds_rdd_4).collect())
#returns all rdd values for which key is avaliable in one rdd and is not available in the other one
print(ds_rdd_1.subtractByKey(ds_rdd_4).collect())
#returm a cartesian product of two rdds
print(ds_rdd_1.cartesian(ds_rdd_4).collect())
#sort rdd by given function
print(ds_rdd_1.sortBy(lambda x: x[1]).collect())
#sort rdd by Key
print(ds_rdd_1.sortByKey().collect())

#Load csv file to dataframe
df = spark.read.format("csv") \
  .option("inferSchema", "true") \
  .option("header", "true") \
  .option("sep", ",") \
  .load("/FileStore/tables/worldcities.csv")

display(df)