<commit_before>from pyspark import SparkContext, SparkConf
import os


conf = SparkConf().setAppName("test")
sc = SparkContext(conf=conf)
_dir = os.path.dirname(os.path.dirname(__file__))
input_file = sc.textFile(_dir + '/input')

node_contrib = (
    input_file
    .map(lambda x: x.split(': '))
    .map(lambda t: (t[0], t[1].split(', ')))  # (a, (b, c))
    .flatMap(
        lambda node: (
            (node[0], (to, 1/len(node[1]))) for to in node[1]
            )
        )
    )  # (node, (to, contribution))
node_contrib = node_contrib.map(lambda op: (op[1][0], (op[0], op[1][1])))  # (to, (from, contrib))

node_rank = (
    input_file
    .map(lambda x: (x.split(": ")[0], 1))
)  # (node, rank)
# ###############################
for i in range(100):
    node_rank.leftOutterJoin(node_contrib)<commit_msg>finish solving the pageRank problem.<commit_after>from pyspark import SparkContext, SparkConf
import os


conf = SparkConf().setAppName("test")
sc = SparkContext(conf=conf)
_dir = os.path.dirname(os.path.dirname(__file__))
input_file = sc.textFile(_dir + '/input')

lines = (
    input_file
    .map(lambda l: l.split(": "))
    .map(lambda l: (l[0], l[1].split(", ")))
)  # (a, (b,c,d))

page_rank = (
    lines
    .map(lambda l: (l[0], (1.0, len(l[1]))))  # (Page, (Rank, NumOfOutgoingCons))->(Page, Contribution)
)
for i in range(3):
    print("we are in iteration number: " + str(i))
    page_contribution = page_rank.map(lambda p: (p[0], (p[1][0]/p[1][1])))  # -> (page, contribution)
    contribution_rdd = (
        page_contribution
        .leftOuterJoin(lines)  # (page, (Contribution, a,b,c,d,...))
        .flatMap(lambda con: ((con[0], (to, con[1][0])) for to in con[1][1]))
        .map(lambda con: (con[1][0], con[1][1]))
    )  # (to, contribution)
    page_rank = (
        page_rank
        .join(contribution_rdd)  # (Page, ((Rank, NumOfOutgoingConnections), contrib, contrib,..))
        .map(lambda x: ((x[0], x[1][0]), x[1][1]))
        .reduceByKey(lambda a, b: a+b)
        .map(lambda pr: (pr[0][0], (pr[1]*0.85 + 0.15, pr[0][1][1])))
    )
print("################################################")
print(page_rank.collect())
page_rank.saveAsTextFile("output")
{'fraction_non_alphanumeric': 0.23772727272727273, 'fraction_numerical': 0.024545454545454544, 'mean_word_length': 3.594989561586639, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7043732', 'n_tokens_mistral': 859, 'n_tokens_neox': 834, 'n_words': 169}