# --------------------------------------------------------
#           PYTHON PROGRAM
# Here is where we are going to define our set of...
# - Imports
# - Global Variables
# - Functions
# ...to achieve the functionality required.
# When executing > python 'this_file'.py in a terminal,
# the Python interpreter will load our program,
# but it will execute nothing yet.
# --------------------------------------------------------

import pyspark

# ------------------------------------------
# FUNCTION combineByKey_lambda
# ------------------------------------------
def combineByKey_lambda(sc):
    # 1. Operation C1: Creation 'parallelize', so as to store the content of the collection [(1, 2), (3, 8), (1, 4), (3, 4), (3, 6)] into an RDD.

    #         C1: parallelize
    # dataset -----------------> inputRDD

    inputRDD = sc.parallelize([(1, 2), (3, 8), (1, 4), (3, 4), (3, 6)])

    # 2. Operation T1: Transformation 'combineByKey', so as to get a new RDD ('combinedRDD') from inputRDD.

    # combineByKey is a higher-order function.
    # It requires as arguments 3 functions:

    # F1: To be applied in parallel to each node of the cluster.
    # The function is responsible of answering this question:
    # How do you want Spark to process the first (key, value) pair for each key k ?

    # If a node contains 1000 entries (key, value) with key 'k', F1 will only be applied once, for the first (key, value) found.
    # F1 must receive as input 1 parameter: The value of the (key, value) pair.
    # F1 must produce as an output 1 parameter: The accumulator accum generated for the pair (key, accum), created after
    # processing the first (key, value).

    # F2: To be applied in parallel to each node of the cluster.
    # The function is responsible of answering this question:
    # How do you want Spark to process all (key, value) pairs for each key k after having processed the first one and have got an accumulator ?

    # If a node contains 1000 entries (key, value) with key 'k', F2 will be applied 999 times, for all except the first (key, value) found.
    # F2 must receive as input 2 parameters:
    # - The accumulor generated until now.
    # - The value of the new (key, value) pair being found.
    # F2 must produce as an output 1 parameter: The updated accumulator, after aggregating it with the new (key, value) being found.

    # F3: To be applied as a whole single process through all nodes of the cluster.
    # The function is responsible of answering this question:
    # How do you want Spark to process all (key, accumulator) pairs so as to get a whole single (key, accumulator) pair ?

    # If combineByKey is applied to n nodes, F3 will be applied n-1 times, to merge all accumulators under a single accumulator.
    # F3 must receive as input 2 parameters:
    # - The meta-accumulor generated until now.
    # - The accumulator generated by node i, being processed now.
    # F3 must produce as an output 1 parameter: The updated accumulator, after aggregating it with the new (key, accumulator) being found.

    #         C1: parallelize             T1: combineByKey
    # dataset -----------------> inputRDD -----------------> combinedRDD

    combinedRDD = inputRDD.combineByKey(lambda x: (x, 1),
                                        lambda x, y: (x[0] + y, x[1] + 1),
                                        lambda x, y: (x[0] + y[0], x[1] + y[1])
                                       )


    # 3. Operation A1: 'collect'.

    #         C1: parallelize             T1: combineByKey               A1: collect
    # dataset -----------------> inputRDD -----------------> combinedRDD ------------> resVAL

    resVAL = combinedRDD.collect()

    # 4. We print by the screen the collection computed in resVAL
    for item in resVAL:
        print(item)


# ------------------------------------------
# FUNCTION createCombiner
# ------------------------------------------
def createCombiner(value):
    # 1. We create the output variable
    res = (value, 1)

    # 2. We return res
    return res


# ------------------------------------------
# FUNCTION mergeValue
# ------------------------------------------
def mergeValue(accum, new_value):
    # 1. We create the output variable
    res = (0, 0)

    # 2. We assign res to the proper value
    val1 = accum[0] + new_value
    val2 = accum[1] + 1
    res = (val1, val2)

    # 3. We return res
    return res


# ------------------------------------------
# FUNCTION mergeCombiners
# ------------------------------------------
def mergeCombiners(accum1, accum2):
    # 1. We create the output variable
    res = (0, 0)

    # 2. We assign res to the proper value
    val1 = accum1[0] + accum2[0]
    val2 = accum1[1] + accum2[1]
    res = (val1, val2)

    # 3. We return res
    return res

# ------------------------------------------
# FUNCTION combineByKey_explicit_function
# ------------------------------------------
def combineByKey_explicit_function(sc):
    # 1. Operation C1: Creation 'parallelize', so as to store the content of the collection [(1, 2), (3, 8), (1, 4), (3, 4), (3, 6)] into an RDD.

    #         C1: parallelize
    # dataset -----------------> inputRDD

    inputRDD = sc.parallelize([(1, 2), (3, 8), (1, 4), (3, 4), (3, 6)])

    # 2. Operation T1: Transformation 'combineByKey', so as to get a new RDD ('combinedRDD') from inputRDD.
    combinedRDD = inputRDD.combineByKey(createCombiner,
                                        mergeValue,
                                        mergeCombiners
                                       )

    # 3. Operation A1: 'collect'.

    #         C1: parallelize             T1: combineByKey               A1: collect
    # dataset -----------------> inputRDD -----------------> combinedRDD ------------> resVAL

    resVAL = combinedRDD.collect()

    # 4. We print by the screen the collection computed in resVAL
    for item in resVAL:
        print(item)



# ------------------------------------------
# FUNCTION my_main
# ------------------------------------------
def my_main(sc):
    print("\n\n--- [BLOCK 1] combineByKey with F1, F2, F3 defined via a lambda expressions ---")
    combineByKey_lambda(sc)

    print("\n\n--- [BLOCK 2] combineByKey with F1, F2, F3 defined via a explicit functions ---")
    combineByKey_explicit_function(sc)


# ---------------------------------------------------------------
#           PYTHON EXECUTION
# This is the main entry point to the execution of our program.
# It provides a call to the 'main function' defined in our
# Python program, making the Python interpreter to trigger
# its execution.
# ---------------------------------------------------------------
if __name__ == '__main__':
    # 1. We configure the Spark Context
    sc = pyspark.SparkContext.getOrCreate()
    sc.setLogLevel('WARN')
    print("\n\n\n")

    # 2. We call to my_main
    my_main(sc)
