<jupyter_start><jupyter_text>PageRank in [Apache Spark](http://spark.apache.org)This is one of the basic examples how Apache Spark works and how it looks like.Make sure to read about [Transformations](http://spark.apache.org/docs/latest/programming-guide.htmltransformations) and especially map, flatMap, join, ...Read more about the RDD's [Python API](http://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD).When using the "Apache Spark" themes kernels in SageMathCloud,the object "`sc`" for the "Spark Context" is already pre-initialized.<jupyter_code>sc<jupyter_output><empty_output><jupyter_text>The data for your simplified link graph.<jupyter_code>link_data = {
    0: [1, 11],
    1: [2, 6],
    2: [1, 0],
    3: [1, 0],
    4: [1],
    5: [0, 1],
    6: [0, 7],
    7: [0, 1, 2, 3, 9],
    8: [0, 5, 9],
    9: [7, 11],
    10: [4, 7],
    11: [0, 10]
}
import networkx as nx
link_graph = nx.DiGraph(link_data)
labels = dict((n, str(n)) for n in link_graph.nodes())
nx.draw_circular(link_graph, labels = labels)<jupyter_output><empty_output><jupyter_text>The initial rank-data is the value 1. for each node.<jupyter_code>ranks = sc.parallelize(link_data.keys()).map(lambda x : (x, 1.))<jupyter_output><empty_output><jupyter_text>This initializes the edges of the graph data as `links`, which are modeled in Spark as key-value-tuples.<jupyter_code>links = sc.parallelize(link_data.items()).cache()<jupyter_output><empty_output><jupyter_text>This is a demonstration of what does happen, when the rank-key-value tuples are joined with the links-key-value tuples.Take a close look, it's a list of tuples in tuples with lists inside of them!<jupyter_code>sorted(links.join(ranks).collect())
def computeContribs(node_urls_rank):
    """
    This function takes elements from the joined dataset above and
    computes the contribution to each outgoing link based on the
    current rank.
    """
    _, (urls, rank) = node_urls_rank
    nb_urls = len(urls)
    for url in urls:
        yield url, rank / nb_urls<jupyter_output><empty_output><jupyter_text>Here is a debug printout to outline what the first operation in the code below is doing:<jupyter_code>c = links.join(ranks).flatMap(computeContribs)
print(c.toDebugString().decode("utf8"))<jupyter_output>(1) PythonRDD[17] at RDD at PythonRDD.scala:43 []
 |  MapPartitionsRDD[16] at mapPartitions at PythonRDD.scala:374 []
 |  ShuffledRDD[15] at partitionBy at NativeMethodAccessorImpl.java:-2 []
 +-(2) PairwiseRDD[14] at _run_module_as_main at /projects/anaconda3/lib/python3.5/runpy.py:170 []
    |  PythonRDD[13] at _run_module_as_main at /projects/anaconda3/lib/python3.5/runpy.py:170 []
    |  UnionRDD[12] at union at NativeMethodAccessorImpl.java:-2 []
    |  PythonRDD[10] at RDD at PythonRDD.scala:43 []
    |  ParallelCollectionRDD[1] at parallelize at PythonRDD.scala:423 []
    |      CachedPartitions: 1; MemorySize: 221.0 B; ExternalBlockStoreSize: 0.0 B; DiskSize: 0.0 B
    |  PythonRDD[11] at RDD at PythonRDD.scala:43 []
    |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:423 []<jupyter_text>Executing the PageRank Algorithm This takes a while to execute. Do    tail -f ~/.smc/jupyter/jupyter-notebook.*    in a Terminal to see what's going on behind the scenes!<jupyter_code>from operator import add

for iteration in range(10):
    # compute contributions of each node where it links to
    contribs = links.join(ranks).flatMap(computeContribs)

    # use a full outer join to make sure, that not well connected nodes aren't dropped
    contribs = links.fullOuterJoin(contribs).mapValues(lambda x : x[1] or 0.0)

    # Sum up all contributions per link
    ranks = contribs.reduceByKey(add)

    # Re-calculate URL ranks
    ranks = ranks.mapValues(lambda rank: rank * 0.85 + 0.15)
    
# Collects all URL ranks
for (link, rank) in sorted(ranks.collect()):
    print("%s has rank: %s." % (link, rank / len(link_data)))<jupyter_output>0 has rank: 0.18868310146544207.
1 has rank: 0.209039646026545.
2 has rank: 0.11721980982628066.
3 has rank: 0.028463192013959495.
4 has rank: 0.037044621083774015.
5 has rank: 0.016041666666666666.
6 has rank: 0.10125661781232116.
7 has rank: 0.09357898007794324.
8 has rank: 0.012499999999999999.
9 has rank: 0.03200485868062616.
10 has rank: 0.05761473312981078.
11 has rank: 0.10655277321663052.<jupyter_text>Comparison with NetworkX<jupyter_code>import networkx as nx
g = nx.DiGraph(link_data)
nx.pagerank(g)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.115273132664437, 'fraction_numerical': 0.07759197324414716, 'mean_word_length': 4.517835178351784, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '501412', 'license': 'Apache-2.0', 'n_tokens_mistral': 1823, 'n_tokens_neox': 1597, 'n_words': 445, 'path': 'data-analysis/spark-pagerank.ipynb', 'repo_name': 'DrXyzzy/cloud-examples', 'size': '4485'}