# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# +
from pyspark.sql import SparkSession

#New API
spark_session = SparkSession\
        .builder\
        .master("spark://192.168.2.113:7077") \
        .appName("marcelloVendruscolo_Assignment3_pA")\
        .config("spark.dynamicAllocation.enabled", True)\
        .config("spark.shuffle.service.enabled", True)\
        .config("spark.dynamicAllocation.executorIdleTimeout","30s")\
        .config("spark.executor.cores",2)\
        .config("spark.driver.port",9998)\
        .config("spark.blockManager.port",10005)\
        .getOrCreate()

#Old API (RDD)
spark_context = spark_session.sparkContext
spark_context.setLogLevel("INFO")


# -

#Helper function to (i) lowercase and (ii) tokenize (split on space) text.
def func_lowercase_split(rdd):
    return rdd.lower().split(' ')


#A.1.1 and A.1.4 - Read the English transcripts with Spark, and count the number of lines and partitions.
print("File: europarl-v7.sv-en.en\n")
en_1 = spark_context.textFile("hdfs://192.168.2.113:9000/europarl/europarl-v7.sv-en.en")
lineCount_en_1 = en_1.count()
print("Line countig: " + str(lineCount_en_1))
print("Partition counting: " + str(en_1.getNumPartitions()))

#A.1.2 and A.1.4 - Read the Swedish transcripts with Spark, and count the number of lines and partitions.
print("File: europarl-v7.sv-en.sv\n")
sv_1 = spark_context.textFile("hdfs://192.168.2.113:9000/europarl/europarl-v7.sv-en.sv")
lineCount_sv_1 = sv_1.count()
print("Line counting: " + str(lineCount_sv_1))
print("Partition counting: " + str(sv_1.getNumPartitions()))

#A.1.3 - Verify that the line counts are the same for the two languages.
print("The line counts are the same for both europarl-v7.sv-en.en and europarl-v7.sv-en.sv: " + str(lineCount_en_1 == lineCount_sv_1))

#A.2.1 - Preprocess the text from both RDDs by lowercase-ing and tokenize-ing (split on space) the text:
en_2 = en_1.map(func_lowercase_split)
sv_2 = sv_1.map(func_lowercase_split)

#A.2.2 - Inspect 10 entries from each of your RDDs to verify your pre-processing.
print("10 entries from the English corpus after pre-processing:\n")
print(en_2.take(10))
print("\n10 entries from the Swedish corpus after pre-processing:\n")
print(sv_2.take(10))

#A.2.3 Verify that the line counts still match after the pre-processing.
lineCount_en_2 = en_2.count()
lineCount_sv_2 = sv_2.count()
print("The line counts are the same for europarl-v7.sv-en.en before and after processing: " + str(lineCount_en_1 == lineCount_en_2))
print("The line counts are the same for europarl-v7.sv-en.sv before and after processing: " + str(lineCount_sv_1 == lineCount_sv_2))
print("The line counts are the same for both europarl-v7.sv-en.en and europarl-v7.sv-en.sv after processing: " + str(lineCount_en_2 == lineCount_sv_2))

#A.3.1 - Use Spark to compute the 10 most frequently according words in the English and Swedish language corpus.
print("The 10 most frequent words in the English corpus:\n")
print(en_2.flatMap(lambda x: x).map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y).takeOrdered(10, key = lambda x: -x[1]))
print("\nThe 10 most frequent words in the Swedish corpus:\n")
print(sv_2.flatMap(lambda x: x).map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y).takeOrdered(10, key = lambda x: -x[1]))

#A.4.1 - Use this parallel corpus to mine some translations in the form of word pairs, for the two languages.
en_3 = en_2.zipWithIndex().map(lambda x: (x[1],x[0]))
sv_3 = sv_2.zipWithIndex().map(lambda x: (x[1],x[0]))
en_sv_0 = en_3.join(sv_3)
en_sv_1 = en_sv_0.filter(lambda x: (not x[1][0] is None) and (not x[1][1] is None)) #line pairs that have an empty/missing “corresponding” sentence.
en_sv_2 = en_sv_1.filter(lambda x: ((len(x[1][0]) <= 15) and (len(x[1][1]) <= 15)))
en_sv_3 = en_sv_2.filter(lambda x: ((len(x[1][0]) >= 2) and (len(x[1][1]) >= 2))) #filter out sentences that are too short.
en_sv_4 = en_sv_3.filter(lambda x: (len(x[1][0]) == len(x[1][1])))
en_sv = en_sv_4.map(lambda x: list(zip(x[1][0],x[1][1]))).flatMap(lambda x: x)
print("Some of the most frequently occurring pairs of words:\n")
print(en_sv.map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y).takeOrdered(20, key = lambda x: -x[1]))

#Release the cores for another application!
spark_context.stop()

StarcoderdataJupyterScriptsDedupFiltered